Copilot commented on code in PR #23004:
URL: https://github.com/apache/kafka/pull/23004#discussion_r3977169868
##########
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 to
+ * {@code PENDING_SHUTDOWN}; 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) {
log.info("Informed to shut down");
final State oldState = setState(State.PENDING_SHUTDOWN);
+ if (oldState == null) {
+ // Shutdown was already requested by another caller (a concurrent
removal, thread
+ // replacement, or client close); that caller owns this thread's
death.
+ return false;
Review Comment:
Returning before updating `leaveGroupRequested` makes a concurrent
`KafkaStreams.close(CloseOptions)` silently ignore its requested membership
operation whenever removal or replacement has already moved the thread to
`PENDING_SHUTDOWN`. For example, replacement can win with `DEFAULT`, then a
user close with `LEAVE_GROUP` gets `false`, and the classic consumer remains in
the group despite the public close option; the previous implementation still
recorded the later operation. Please preserve the ownership result needed by
replacement while providing a way for the close path to apply its explicit
operation (for example, a separate operation-update path or caller-aware
shutdown API).
--
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]
