KAFKA-5052; Don't pass underlying internal exception to RetriableCommitFailedException

Author: Apurva Mehta <apurva@confluent.io>

Reviewers: Ismael Juma <ismael@juma.me.uk>

Closes #2838 from apurvam/KAFKA-5052-dont-send-uderlying-exception-to-retriableoffsetcommitfailed

clients/src/main/java/org/apache/kafka/clients/consumer/RetriableCommitFailedException.java

@@ -27,7 +27,7 @@ public class RetriableCommitFailedException extends RetriableException {
     }
 
     public RetriableCommitFailedException(String message) {
-        super(message);
+        super("Offset commit failed with a retriable exception. You should retry committing offsets. The underlying error was: " + message);
     }
 
     public RetriableCommitFailedException(String message, Throwable t) {

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java

@@ -521,7 +521,7 @@ public final class ConsumerCoordinator extends AbstractCoordinator {
                 @Override
                 public void onFailure(RuntimeException e) {
                     pendingAsyncCommits.decrementAndGet();
-                    completedOffsetCommits.add(new OffsetCommitCompletion(callback, offsets, new RetriableCommitFailedException(e)));
+                    completedOffsetCommits.add(new OffsetCommitCompletion(callback, offsets, new RetriableCommitFailedException(e.getMessage())));
                 }
             });
         }
@@ -550,7 +550,7 @@ public final class ConsumerCoordinator extends AbstractCoordinator {
                 Exception commitException = e;
 
                 if (e instanceof RetriableException)
-                    commitException = new RetriableCommitFailedException(e);
+                    commitException = new RetriableCommitFailedException(e.getMessage());
 
                 completedOffsetCommits.add(new OffsetCommitCompletion(cb, offsets, commitException));
             }