lucasbru commented on code in PR #22951:
URL: https://github.com/apache/kafka/pull/22951#discussion_r3961461837
##########
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:
One edge case: registerStateStores() locks before adding any store, so if
init fails right after locking (stores stays empty), the lock is left held. If
a different thread later closes this task, it takes this early-return path - no
lock() attempt, no WARN, and unlock() is a no-op since it's not the recorded
owner. So we'd silently lose the lock forever with no log trail, whereas before
we'd at least have gotten the same WARN this PR is trying to remove. Maybe
worth a log line in this branch too?
##########
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:
recycle() doesn't clear the `stores` map, only close() does. So a stateful
task that's recycled during hand-off would still have hasRegisteredStores() ==
true here, and this guard wouldn't fire for it. Doesn't that mean the WARN
would still show up for the stateful hand-off case, and this only really helps
stateless tasks? Also, since we return before calling stateMgr.close(), we now
skip maybeDowngradeOffsets() unconditionally whenever a task has zero
registered stores - is that ok during a downgrade to an old upgrade.from?
--
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]
