Copilot commented on code in PR #23004:
URL: https://github.com/apache/kafka/pull/23004#discussion_r3982316056
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -377,6 +377,10 @@ public boolean isStartingRunningOrPartitionAssigned() {
private final AtomicLong cacheResizeSize = new AtomicLong(-1L);
private final
AtomicReference<org.apache.kafka.streams.CloseOptions.GroupMembershipOperation>
leaveGroupRequested =
new
AtomicReference<>(org.apache.kafka.streams.CloseOptions.GroupMembershipOperation.DEFAULT);
+ // Guards the hand-off of leaveGroupRequested to completeShutdown: an
update is applied if and
+ // only if it acquires the lock before the shutting-down thread consumes
the operation.
+ private final Object leaveGroupRequestedLock = new Object();
+ private boolean leaveGroupRequestedConsumed = false;
Review Comment:
This lock does not guard every write to the value it protects: `run()` still
calls `leaveGroupRequested.set(LEAVE_GROUP)` directly (line 961). If
`close(...REMAIN_IN_GROUP/DEFAULT)` wins `shutdown()` and the worker then
enters the exception path, that unsynchronized write can overwrite the explicit
close operation before `completeShutdown()` consumes it, contradicting the new
hand-off guarantee. Route the exception-path write through the same protocol
and preserve an operation whose shutdown was already initiated by another
caller.
--
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]
