Repository: apache/kafka
File: streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java (Java)

Code before:
            }
        } catch (final RuntimeException swallow) {
            log.warn("Error removing dirty task {}. " +
                "Since the task is closing dirty, the following exception is swallowed: {}",
                task.id(), swallow);
        }
    }

    private void closeTaskClean(final Task task) {
        task.closeClean();
        tasks.removeTask(task);
    }

    void shutdown(final boolean clean) {
        shutdownStateUpdater();
        shutdownSchedulingTaskManager();

        final AtomicReference<RuntimeException> firstException = new AtomicReference<>(null);

        // TODO: change type to `StreamTask`
        final Set<Task> activeTasks = new TreeSet<>(Comparator.comparing(Task::id));
        activeTasks.addAll(tasks.activeTasks());
        final Set<Task> standbyTasks = new TreeSet<>(Comparator.comparing(Task::id));
        standbyTasks.addAll(tasks.standbyTasks());

        Set<Task> pendingActiveTasks = tasks.drainPendingActiveTasksToInit();
        activeTasks.addAll(pendingActiveTasks);
        tasks.addPendingTasksToClose(pendingActiveTasks);
        Set<Task> pendingStandbyTasks = tasks.drainPendingStandbyTasksToInit();
        standbyTasks.addAll(pendingStandbyTasks);
        tasks.addPendingTasksToClose(pendingStandbyTasks);

        executeAndMaybeSwallow(
            clean,
            () -> closeAndCleanUpTasks(activeTasks, standbyTasks, clean),
            e -> firstException.compareAndSet(null, e),
            e -> log.warn("Ignoring an exception while unlocking remaining task directories.", e)
        );

        executeAndMaybeSwallow(
            clean,
            activeTaskCreator::close,
            e -> firstException.compareAndSet(null, e),
            e -> log.warn("Ignoring an exception while closing thread producer.", e)
        );

        tasks.clear();

        // this should be called after closing all tasks and clearing them from `tasks` to make sure we unlock the dir
        // for any tasks that may have still been in CREATED at the time of shutdown, since Task#close will not do so
        executeAndMaybeSwallow(

Review comment (bug):
Seems this should be `final` -- wondering why the build did not fail with a checktsyle error? Can we check this?

Code after:
            }
        } catch (final RuntimeException swallow) {
            log.warn("Error removing dirty task {}. " +
                "Since the task is closing dirty, the following exception is swallowed: {}",
                task.id(), swallow);
        }
    }

    private void closeTaskClean(final Task task) {
        task.closeClean();
        tasks.removeTask(task);
    }

    void shutdown(final boolean clean) {
        shutdownStateUpdater();
        shutdownSchedulingTaskManager();

        final AtomicReference<RuntimeException> firstException = new AtomicReference<>(null);

        // TODO: change type to `StreamTask`
        final Set<Task> activeTasks = new TreeSet<>(Comparator.comparing(Task::id));
        activeTasks.addAll(tasks.activeTasks());
        // TODO: change type to `StandbyTask`
        final Set<Task> standbyTasks = new TreeSet<>(Comparator.comparing(Task::id));
        standbyTasks.addAll(tasks.standbyTasks());

        final Set<Task> pendingActiveTasks = tasks.drainPendingActiveTasksToInit();
        activeTasks.addAll(pendingActiveTasks);
        tasks.addPendingTasksToClose(pendingActiveTasks);
        final Set<Task> pendingStandbyTasks = tasks.drainPendingStandbyTasksToInit();
        standbyTasks.addAll(pendingStandbyTasks);
        tasks.addPendingTasksToClose(pendingStandbyTasks);

        executeAndMaybeSwallow(
            clean,
            () -> closeAndCleanUpTasks(activeTasks, standbyTasks, clean),
            e -> firstException.compareAndSet(null, e),
            e -> log.warn("Ignoring an exception while unlocking remaining task directories.", e)
        );

        executeAndMaybeSwallow(
            clean,
            activeTaskCreator::close,
            e -> firstException.compareAndSet(null, e),
            e -> log.warn("Ignoring an exception while closing thread producer.", e)
        );

        tasks.clear();

        // this should be called after closing all tasks and clearing them from `tasks` to make sure we unlock the dir
        // for any tasks that may have still been in CREATED at the time of shutdown, since Task#close will not do so