private void tryToLockAllNonEmptyTaskDirectories() {
        // Always clear the set at the beginning as we're always dealing with the
        // current set of actually-locked tasks.
        lockedTaskDirectories.clear();

        final Map<TaskId, Task> allTasks = allTasks();
        for (final TaskDirectory taskDir : stateDirectory.listNonEmptyTaskDirectories()) {
            final File dir = taskDir.file();
            final String namedTopology = taskDir.namedTopology();
            try {
                final TaskId id = parseTaskDirectoryName(dir.getName(), namedTopology);
                if (stateDirectory.lock(id)) {
                    // Check again in case the cleaner thread ran and emptied the directory
                    if (stateDirectory.directoryForTaskIsEmpty(id)) {
                        log.debug("Releasing lock on empty directory for task {}", id);
                        stateDirectory.unlock(id);
                    } else {
                        lockedTaskDirectories.add(id);
                        if (!allTasks.containsKey(id)) {
                            log.debug("Temporarily locked unassigned task {} for the upcoming rebalance", id);
                        }
                    }
                }
            } catch (final TaskIdFormatException e) {
                // ignore any unknown files that sit in the same directory
            }
        }
    }
Makes a weak attempt to lock all non-empty task directories in the state dir. We are responsible for computing and
reporting the offset sum for any unassigned tasks we obtain the lock for in the upcoming rebalance. Tasks
that we locked but didn't own will be released at the end of the rebalance (unless of course we were
assigned the task as a result of the rebalance). This method should be idempotent.
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java