cadonna commented on code in PR #17116:
URL: https://github.com/apache/kafka/pull/17116#discussion_r1760880614
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java:
##########
@@ -93,21 +93,24 @@ static void registerStateStores(final Logger log,
}
final TaskId id = stateMgr.taskId();
- if (!stateDirectory.lock(id)) {
+ if (!stateDirectory.canTryLock(id, System.currentTimeMillis())) {
+ log.trace("Task {} is still not allowed to retry acquiring the
state directory lock", id);
Review Comment:
I also wanted to propose something similar what Lucas wrote. I think that is
simpler. But why not have a private method `canTryInitializeIfNeeded()` in
`TaskManager` and create the backoff record when the `LockException
lockException` is thrown? The initialization is the actual call we want to
backoff, right?
--
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]
