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

Code before:
            }
        }

        private void runOnce() throws InterruptedException {
            performActionsOnTasks();
            restoreTasks();
            checkAllUpdatingTaskStates(time.milliseconds());
            waitIfAllChangelogsCompletelyRead();
        }

        private void performActionsOnTasks() {
            while (!tasksAndActions.isEmpty()) {
                Task toInitialize = null;
                tasksAndActionsLock.lock();
                try {
                    final TaskAndAction taskAndAction = tasksAndActions.remove();
                    final Action action = taskAndAction.getAction();
                    switch (action) {
                        case ADD:
                            final Task task = taskAndAction.getTask();
                            final Task existingTask = updatingTasks.putIfAbsent(task.id(), task);
                            if (existingTask != null) {
                                throw new IllegalStateException((existingTask.isActive() ? "Active" : "Standby") + " task " + task.id() + " already exist, " +
                                    "should not try to add another " + (task.isActive() ? "active" : "standby") + " task with the same id. " + BUG_ERROR_MESSAGE);
                            }
                            toInitialize = task;
                            break;
                        case REMOVE:
                            removeTask(taskAndAction.getTaskId());
                            break;
                    }
                } finally {
                    tasksAndActionsLock.unlock();
                }
                if (toInitialize != null) {
                    initializeTask(toInitialize);
                }
            }
        }

        private void restoreTasks() {
            try {
                changelogReader.restore(updatingTasks);
            } catch (final TaskCorruptedException taskCorruptedException) {
                handleTaskCorruptedException(taskCorruptedException);
            } catch (final StreamsException streamsException) {
                handleStreamsException(streamsException);
            }
            final Set<TopicPartition> completedChangelogs = changelogReader.completedChangelogs();
            final List<Task> activeTasks = updatingTasks.values().stream().filter(Task::isActive).collect(Collectors.toList());
            for (final Task task : activeTasks) {

Review comment (refactor):
Could you extract this code to a method that returns a boolean that can be assigned to `toInitialize`? I think that makes the code more readable.

Code after:
            }
        }

        private void runOnce() throws InterruptedException {
            performActionsOnTasks();
            restoreTasks();
            checkAllUpdatingTaskStates(time.milliseconds());
            waitIfAllChangelogsCompletelyRead();
        }

        private void performActionsOnTasks() {
            while (!tasksAndActions.isEmpty()) {
                Task toInitialize = null;
                tasksAndActionsLock.lock();
                try {
                    final TaskAndAction taskAndAction = tasksAndActions.remove();
                    final Action action = taskAndAction.getAction();
                    switch (action) {
                        case ADD:
                            addTask(taskAndAction.getTask());
                            toInitialize = taskAndAction.getTask();
                            break;
                        case REMOVE:
                            removeTask(taskAndAction.getTaskId());
                            break;
                    }
                } finally {
                    tasksAndActionsLock.unlock();
                }
                if (toInitialize != null) {
                    initializeTask(toInitialize);
                }
            }
        }

        private void restoreTasks() {
            try {
                changelogReader.restore(updatingTasks);
            } catch (final TaskCorruptedException taskCorruptedException) {
                handleTaskCorruptedException(taskCorruptedException);
            } catch (final StreamsException streamsException) {
                handleStreamsException(streamsException);
            }
            final Set<TopicPartition> completedChangelogs = changelogReader.completedChangelogs();
            final List<Task> activeTasks = updatingTasks.values().stream().filter(Task::isActive).collect(Collectors.toList());
            for (final Task task : activeTasks) {
                maybeCompleteRestoration((StreamTask) task, completedChangelogs);
            }
        }

        private void handleRuntimeException(final RuntimeException runtimeException) {