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

Code before:
            return updatingTasks.values().stream()
                .filter(t -> !t.isActive())
                .count();
        }

        public long getNumRestoringActiveTasks() {
            return updatingTasks.values().stream()
                .filter(Task::isActive)
                .count();
        }

        public long getNumPausedStandbyTasks() {
            return pausedTasks.values().stream()
                .filter(t -> !t.isActive())
                .count();
        }

        public long getNumPausedActiveTasks() {
            return pausedTasks.values().stream()
                .filter(Task::isActive)
                .count();
        }

        @Override
        public void run() {
            log.info("State updater thread started");
            try {
                while (isRunning.get()) {
                    try {
                        runOnce();
                    } catch (final InterruptedException | InterruptException interruptedException) {
                        return;
                    }
                }
            } catch (final RuntimeException anyOtherException) {
                handleRuntimeException(anyOtherException);
            } finally {
                Thread.interrupted(); // Clear the interrupted flag.
                removeAddedTasksFromInputQueue();
                removeUpdatingAndPausedTasks();
                updaterMetrics.clear();
                shutdownGate.countDown();
                log.info("State updater thread shutdown");
            }
        }

        // In each iteration:
        //   1) check if updating tasks need to be paused
        //   2) check if paused tasks need to be resumed
        //   3) restore those updating tasks
        //   4) checkpoint those updating task states

Review comment (suggestion):
We can get rid of this comment since `runOnce()` does not throw a `InterruptedException` anymore.

Code after:
            return updatingTasks.values().stream()
                .filter(Task::isActive)
                .count();
        }

        public long getNumPausedStandbyTasks() {
            return pausedTasks.values().stream()
                .filter(t -> !t.isActive())
                .count();
        }

        public long getNumPausedActiveTasks() {
            return pausedTasks.values().stream()
                .filter(Task::isActive)
                .count();
        }

        @Override
        public void run() {
            log.info("State updater thread started");
            try {
                while (isRunning.get()) {
                    runOnce();
                }
            } catch (final RuntimeException anyOtherException) {
                handleRuntimeException(anyOtherException);
            } finally {
                removeAddedTasksFromInputQueue();
                removeUpdatingAndPausedTasks();
                updaterMetrics.clear();
                shutdownGate.countDown();
                log.info("State updater thread shutdown");
            }
        }

        // In each iteration:
        //   1) check if updating tasks need to be paused
        //   2) check if paused tasks need to be resumed
        //   3) restore those updating tasks
        //   4) checkpoint those updating task states
        //   5) idle waiting if there is no more tasks to be restored
        //
        //   Note that, 1-3) are measured as restoring time, while 4) and 5) measured separately
        //   as checkpointing time and idle time
        private void runOnce() {
            final long totalStartTimeMs = time.milliseconds();
            performActionsOnTasks();

            resumeTasks();
            pauseTasks();
            restoreTasks(totalStartTimeMs);