Repository: kafka Updated Branches: refs/heads/trunk 23b8740d8 -> e4ef8e664
KAFKA-3122; Fix memory leak in `Sender.completeBatch` on TOPIC_AUTHORIZATION_FAILED Also fix missing call to `sensors.record` on this error. Author: Ismael Juma <[email protected]> Reviewers: Jason Gustafson, Guozhang Wang Closes #791 from ijuma/fix-producer-memory-leak-on-authorization-exception Project: http://git-wip-us.apache.org/repos/asf/kafka/repo Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/e4ef8e66 Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/e4ef8e66 Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/e4ef8e66 Branch: refs/heads/trunk Commit: e4ef8e6640a266da1d9135282ae41d18f44f8025 Parents: 23b8740 Author: Ismael Juma <[email protected]> Authored: Tue Jan 19 22:50:24 2016 -0800 Committer: Guozhang Wang <[email protected]> Committed: Tue Jan 19 22:50:24 2016 -0800 ---------------------------------------------------------------------- .../org/apache/kafka/clients/producer/internals/Sender.java | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/kafka/blob/e4ef8e66/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java index b8215e1..aa30716 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java @@ -289,11 +289,14 @@ public class Sender implements Runnable { error); this.accumulator.reenqueue(batch, now); this.sensors.recordRetries(batch.topicPartition.topic(), batch.recordCount); - } else if (error == Errors.TOPIC_AUTHORIZATION_FAILED) { - batch.done(baseOffset, new TopicAuthorizationException(batch.topicPartition.topic())); } else { + RuntimeException exception; + if (error == Errors.TOPIC_AUTHORIZATION_FAILED) + exception = new TopicAuthorizationException(batch.topicPartition.topic()); + else + exception = error.exception(); // tell the user the result of their request - batch.done(baseOffset, error.exception()); + batch.done(baseOffset, exception); this.accumulator.deallocate(batch); if (error != Errors.NONE) this.sensors.recordErrors(batch.topicPartition.topic(), batch.recordCount);
