Copilot commented on code in PR #23004:
URL: https://github.com/apache/kafka/pull/23004#discussion_r3979082726
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -1926,15 +1921,29 @@ 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
+ * (use {@link #updateGroupMembershipOperation} to change it)
*/
- 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;
+ }
leaveGroupRequested.set(operation);
if (oldState == State.CREATED) {
// The thread may not have been started. Take responsibility for
shutting down
completeShutdown(true);
}
+ return true;
+ }
+
+ public void updateGroupMembershipOperation(final
org.apache.kafka.streams.CloseOptions.GroupMembershipOperation operation) {
+ leaveGroupRequested.set(operation);
Review Comment:
This update is not coordinated with `completeShutdown()`, which reads
`leaveGroupRequested` before closing the consumer. After `shutdown(operation)`
returns `false`, the target thread can read the old operation before this
method executes, so a concurrent client close can still silently lose its
requested group-membership behavior. Please make the “already pending +
override operation” action atomic with the operation consumed during shutdown
(rather than splitting it across `shutdown` and this setter).
--
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]
