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

…itFailedException Author: Apurva Mehta <[email protected]> Reviewers: Ismael Juma <[email protected]> Closes apache#2838 from apurvam/KAFKA-5052-dont-send-uderlying-exception-to-retriableoffsetcommitfailed
LioRoger · Apr 11, 2017 · 749e9e1 · 749e9e1
1 parent c31958e
commit 749e9e1
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 3 deletions.
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/RetriableCommitFailedException.java b/clients/src/main/java/org/apache/kafka/clients/consumer/RetriableCommitFailedException.java
@@ -27,7 +27,7 @@ public RetriableCommitFailedException(Throwable t) {
     }
 
     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) {

diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
@@ -521,7 +521,7 @@ public void onSuccess(Void value) {
                 @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 void onFailure(RuntimeException e) {
                 Exception commitException = e;
 
                 if (e instanceof RetriableException)
-                    commitException = new RetriableCommitFailedException(e);
+                    commitException = new RetriableCommitFailedException(e.getMessage());
 
                 completedOffsetCommits.add(new OffsetCommitCompletion(cb, offsets, commitException));
             }