lucasbru commented on code in PR #23004:
URL: https://github.com/apache/kafka/pull/23004#discussion_r3977373985
##########
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:
Looks valid to me.
--
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]
