You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by gu...@apache.org on 2016/01/20 07:50:28 UTC
kafka git commit: KAFKA-3122;
Fix memory leak in `Sender.completeBatch` on
TOPIC_AUTHORIZATION_FAILED
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 <is...@juma.me.uk>
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 <is...@juma.me.uk>
Authored: Tue Jan 19 22:50:24 2016 -0800
Committer: Guozhang Wang <wa...@gmail.com>
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);