nileshkumar3 commented on code in PR #22951:
URL: https://github.com/apache/kafka/pull/22951#discussion_r3941644966
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java:
##########
@@ -171,6 +171,16 @@ static void closeStateManager(final Logger log,
final TaskId id = stateMgr.taskId();
log.trace("Closing state manager for {} task {}", taskType, id);
+ // Nothing to close: the state manager has no registered stores (e.g.
they were already
+ // closed during a previous hand-off, or the task is stateless). Skip
acquiring the
+ // per-task state-directory lock so we don't emit a misleading
lock-contention warning
+ // for a benign back-to-back rebalance. We must still run the close
path when we intend
+ // to wipe the state store, since that deletes the on-disk task
directory.
+ if (!wipeStateStore && !stateMgr.hasRegisteredStores()) {
+ log.trace("No registered state stores to close for {} task {};
skipping lock acquisition", taskType, id);
+ return;
+ }
Review Comment:
agreed, this is a real gap. Since `registerStateStores` takes the
state-directory lock before any store is registered, the early return could
leave the lock cached if initialization failed before the first store
registered. Applied as suggested: the lock is now released before returning,
and the test expectations are updated to match.
--
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]
