You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by jg...@apache.org on 2016/09/30 18:50:43 UTC
kafka git commit: KAFKA-4235;
Fix race condition in Sender.initiateClose that may cause missed
callbacks
Repository: kafka
Updated Branches:
refs/heads/trunk 429224300 -> 18a081261
KAFKA-4235; Fix race condition in Sender.initiateClose that may cause missed callbacks
Author: Jiangjie Qin <be...@gmail.com>
Reviewers: Ismael Juma <is...@juma.me.uk>, Joel Koshy <jj...@gmail.com>, Jason Gustafson <ja...@confluent.io>
Closes #1939 from becketqin/KAFKA-4235
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/18a08126
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/18a08126
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/18a08126
Branch: refs/heads/trunk
Commit: 18a0812618a151e8ac700545699173b10caf692b
Parents: 4292243
Author: Jiangjie Qin <be...@gmail.com>
Authored: Fri Sep 30 11:49:37 2016 -0700
Committer: Jason Gustafson <ja...@confluent.io>
Committed: Fri Sep 30 11:49:54 2016 -0700
----------------------------------------------------------------------
.../java/org/apache/kafka/clients/producer/internals/Sender.java | 4 +++-
1 file changed, 3 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/18a08126/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 30f8887..6471dad 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
@@ -239,8 +239,10 @@ public class Sender implements Runnable {
* Start closing the sender (won't actually complete until all data is sent out)
*/
public void initiateClose() {
- this.running = false;
+ // Ensure accumulator is closed first to guarantee that no more appends are accepted after
+ // breaking from the sender loop. Otherwise, we may miss some callbacks when shutting down.
this.accumulator.close();
+ this.running = false;
this.wakeup();
}