frankvicky commented on code in PR #23004:
URL: https://github.com/apache/kafka/pull/23004#discussion_r3987570646
##########
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:
I would take the simplification.
This will make the semantic very clear, the owner of the `PENDING_SHUTDOWN`
transition decides the operation.
--
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]
