private void shutdownSchedulingTaskManager() {
        if (schedulingTaskManager != null) {
            schedulingTaskManager.shutdown(Duration.ofMillis(Long.MAX_VALUE));
        }
    }
We must release the lock for any unassigned tasks that we temporarily locked in preparation for a
rebalance in {@link #tryToLockAllNonEmptyTaskDirectories()}.
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java