mjsax commented on code in PR #23004:
URL: https://github.com/apache/kafka/pull/23004#discussion_r3985223552
##########
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:
My Claude argues this is the right thing to do, because it's the current
`trunk` behavior, of last-close/write wins. -- But I am not sure if I agree or
not.
Just raising it for consideration, not as request to necessarily change it.
-- Do we believe the current `trunk` behavior is actually correct and desired?
It seems unnecessary to me to have (and preserve) this exact behavior, and if
we believe it actually has not real benefit (and does not break anything if we
change it), it seems to be an opportunity to simplify the code? We could remove
`updateGroupMembershipOperation()` entirely and also simplify other code
reading the close-operation (and can also remove the "consumes" boolean).
Thoughts? \cc @lucasbru
--
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]
