Wait for heartbeat thread to terminate in consumer close
Author: Rajini Sivaram
Reviewers: Apurva Mehta Ismael Juma Jason Gustafson
Closes from rajinisivaram/KAFKA-4786
@@ -315,6 +315,19 @@ public abstract class AbstractCoordinator implements Closeable {
heartbeatThread.disable();
}
+ private void closeHeartbeatThread() {
+ if (heartbeatThread != null) {
+ heartbeatThread.close();
+
+ try {
+ heartbeatThread.join();
+ } catch (InterruptedException e) {
+ log.warn("Interrupted while waiting for consumer heartbeat thread to close");
+ throw new InterruptException(e);
+ }
+ }
+ }
+
// visible for testing. Joins the group without starting the heartbeat thread.
void joinGroupIfNeeded() {
while (needRejoin() || rejoinIncomplete()) {
@@ -652,9 +665,14 @@ public abstract class AbstractCoordinator implements Closeable {
close(0);
}
- protected synchronized void close(long timeoutMs) {
- if (heartbeatThread != null)
- heartbeatThread.close();
+ protected void close(long timeoutMs) {
+ try {
+ closeHeartbeatThread();
+ } finally {
+
+ // Synchronize after closing the heartbeat thread since heartbeat thread
+ // needs this lock to complete and terminate after close flag is set.
+ synchronized (this) {
maybeLeaveGroup();
// At this point, there may be pending commits (async commits or sync commits that were
@@ -666,6 +684,8 @@ public abstract class AbstractCoordinator implements Closeable {
log.warn("Close timed out with {} pending requests to coordinator, terminating client connections for group {}.",
client.pendingRequestCount(coordinator), groupId);
}
+ }
+ }
/**
* Leave the current group and reset local generation/memberId.
