frankvicky commented on code in PR #23004:
URL: https://github.com/apache/kafka/pull/23004#discussion_r3948251280
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -1926,15 +1921,24 @@ private long advanceNowAndComputeLatency() {
* (e.g., in testing), hence the state is set only the first time
*
* @param operation the group membership operation to apply on shutdown.
Must be one of LEAVE_GROUP or REMAIN_IN_GROUP.
+ * @return true if this call initiated the shutdown, i.e., transitioned
the thread out of an
+ * alive state; false if the thread was already shutting down or
dead, in which case
+ * the group membership operation of the earlier shutdown request
is kept
*/
- public void shutdown(final
org.apache.kafka.streams.CloseOptions.GroupMembershipOperation operation) {
+ public boolean shutdown(final
org.apache.kafka.streams.CloseOptions.GroupMembershipOperation operation) {
Review Comment:
Both are intentional, and I added a comment at each call site to say so.
- The terminate branch in `addStreamThread` acts on a thread that is still
in `CREATED`. The only other caller that can get there first is `close()`, and
whichever caller wins runs `completeShutdown` itself, so the bookkeeping that
follows is correct either way.
- `shutdownHelper` joins every thread afterwards regardless of who initiated
its shutdown, so there is nothing to do differently on `false`.
--
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]
