private void shutdownStateUpdater() {
        if (stateUpdater != null) {
            final Map<TaskId, CompletableFuture<StateUpdater.RemovedTaskResult>> futures = new LinkedHashMap<>();
            for (final Task task : stateUpdater.getTasks()) {
                final CompletableFuture<StateUpdater.RemovedTaskResult> future = stateUpdater.remove(task.id());
                futures.put(task.id(), future);
            }
            final Set<Task> tasksToCloseClean = new HashSet<>();
            final Set<Task> tasksToCloseDirty = new HashSet<>();
            addToTasksToClose(futures, tasksToCloseClean, tasksToCloseDirty);
            stateUpdater.shutdown(Duration.ofMillis(Long.MAX_VALUE));

            for (final Task task : tasksToCloseClean) {
                tasks.addTask(task);
            }
            for (final Task task : tasksToCloseDirty) {
                closeTaskDirty(task, false);
            }
            for (final StateUpdater.ExceptionAndTask exceptionAndTask : stateUpdater.drainExceptionsAndFailedTasks()) {
                final Task failedTask = exceptionAndTask.task();
                closeTaskDirty(failedTask, false);
            }
        }
    }
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