nileshkumar3 commented on code in PR #22951:
URL: https://github.com/apache/kafka/pull/22951#discussion_r3999973715
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java:
##########
@@ -171,6 +171,20 @@ 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()) {
Review Comment:
Agreed — that split was backwards.
lock() failing only means another thread is (or was) in the map, so it
cannot tell a hand-off from a stuck lock. WARN is now only when lockOwner() is
a terminated thread (never unlocked). A live other thread, or lockOwner() ==
null because they unlocked between the two calls, is DEBUG.
A hung-but-still-alive owner is indistinguishable from a live hand-off here;
leftover locks are still reported at StateDirectory.close().
hasRegisteredStores() was unused after dropping the early-return; removed it.
--
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]
