frankvicky commented on code in PR #23004:
URL: https://github.com/apache/kafka/pull/23004#discussion_r3987581100
##########
streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java:
##########
@@ -1527,9 +1629,20 @@ private Thread shutdownHelper(
// we don't attempt to join it and cause a deadlock
return new Thread(() -> {
// notify all the threads to stop; avoid deadlocks by stopping any
- // further state reports from the thread since we're shutting down
+ // further state reports from the thread since we're shutting down.
+ // A thread may already be shutting down (initiated by a removal
or a thread
+ // replacement), in which case shutdown() returns false. Closing
is an explicit
+ // client-level request whose group membership operation applies
to every thread,
+ // so record it on those threads too; every thread is joined below
regardless of
+ // who initiated its shutdown.
int numStreamThreads = processStreamThread(
- streamThread -> streamThread.shutdown(operation)
+ streamThread -> {
+ if (!streamThread.shutdown(operation)
+ &&
!streamThread.updateGroupMembershipOperation(operation)) {
Review Comment:
We could have a follow-up PR for simplification, if we agree to do it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
