Copilot commented on code in PR #22951:
URL: https://github.com/apache/kafka/pull/22951#discussion_r3913016399
##########
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:
The early return when there are no registered stores skips unlocking the
task’s state-directory lock. If the current thread already holds the lock (eg
via StateManagerUtil.registerStateStores()) but store initialization failed
before any store was registered, this can leave the lock cached in
StateDirectory and block future lock acquisition for the task.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java:
##########
@@ -438,6 +438,15 @@ boolean hasCorruptedStores() {
return stores.values().stream().anyMatch(m -> m.corrupted);
}
+ /**
+ * @return whether this state manager currently has any registered state
stores. After
+ * {@link #close()} the registered stores are cleared, so an empty
result means
+ * there is nothing left to close.
+ */
Review Comment:
Javadoc says “an empty result means …” but this method returns a boolean;
this is confusing/misleading for readers.
--
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]
