ableegoldman commented on code in PR #15088:
URL: https://github.com/apache/kafka/pull/15088#discussion_r1439872897
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1229,10 +1229,21 @@ private void tryToLockAllNonEmptyTaskDirectories() {
final String namedTopology = taskDir.namedTopology();
try {
final TaskId id = parseTaskDirectoryName(dir.getName(),
namedTopology);
- if (stateDirectory.lock(id)) {
- lockedTaskDirectories.add(id);
- if (!allTasks.containsKey(id)) {
- log.debug("Temporarily locked unassigned task {} for
the upcoming rebalance", id);
+ boolean lockedEmptyDirectory = false;
+ try {
+ if (stateDirectory.lock(id)) {
+ if (stateDirectory.directoryForTaskIsEmpty(id)) {
+ lockedEmptyDirectory = true;
Review Comment:
This is maybe a bit pedantic, but I feel like it would be easier to
understand the flow & intent of this code if we just unlocked it directly right
here (instead of introducing the `lockedEmptyDirectory` flag and then unlocking
it later on)
Also, since we log a debug message when we do lock the directory, we might
as well put a similar debug message in this branch too. Something like
`"Released lock on empty directory for task {}"`
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
