nileshkumar3 commented on code in PR #22951:
URL: https://github.com/apache/kafka/pull/22951#discussion_r3968963782
##########
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:
Good catch — recycle() leaves stores populated, so the early return would
not fire on a stateful hand-off.
I dropped this guard and only downgrading the WARN when lock() fails and
lockOwner is another thread in this process. close() would still run when we
get the lock, so maybeDowngradeOffsets() stays as-is. Does that look right to
you?
--
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]
