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

Code before:
            return tasksAndActionsToProcess;
        }

        private void addTask(final Task task) {
            final TaskId taskId = task.id();
            if (isStateless(task)) {
                addToRestoredTasks((StreamTask) task);
                log.info("Stateless active task " + taskId + " was added to the restored tasks of the state updater");
            } else if (topologyMetadata.isPaused(taskId.topologyName())) {
                pausedTasks.put(taskId, task);
                changelogReader.register(task.changelogPartitions(), task.stateManager());
                log.debug((task.isActive() ? "Active" : "Standby")
                    + " task " + taskId + " was directly added to the paused tasks.");
            } else {
                final Task existingTask = updatingTasks.putIfAbsent(taskId, task);
                if (existingTask != null) {
                    throw new IllegalStateException((existingTask.isActive() ? "Active" : "Standby") + " task " + taskId + " already exist, " +
                        "should not try to add another " + (task.isActive() ? "active" : "standby") + " task with the same id. " + BUG_ERROR_MESSAGE);
                }
                changelogReader.register(task.changelogPartitions(), task.stateManager());
                if (task.isActive()) {
                    log.info("Stateful active task " + taskId + " was added to the state updater");
                    changelogReader.enforceRestoreActive();
                } else {
                    log.info("Standby task " + taskId + " was added to the state updater");
                    if (updatingTasks.size() == 1) {
                        changelogReader.transitToUpdateStandby();
                    }
                }
            }
        }

        private void removeTask(final TaskId taskId) {
            final Task task;
            if (updatingTasks.containsKey(taskId)) {
                task = updatingTasks.get(taskId);
                task.maybeCheckpoint(true);
                final Collection<TopicPartition> changelogPartitions = task.changelogPartitions();
                changelogReader.unregister(changelogPartitions);
                removedTasks.add(task);
                updatingTasks.remove(taskId);
                if (task.isActive()) {
                    transitToUpdateStandbysIfOnlyStandbysLeft();
                }
                log.info((task.isActive() ? "Active" : "Standby")
                    + " task " + task.id() + " was removed from the updating tasks and added to the removed tasks.");
            } else if (pausedTasks.containsKey(taskId)) {
                task = pausedTasks.get(taskId);
                final Collection<TopicPartition> changelogPartitions = task.changelogPartitions();
                changelogReader.unregister(changelogPartitions);
                removedTasks.add(task);

Review comment (suggestion):
To be consistent, we should also verify here if the task already exists similar to the `else`-branch.  

Code after:
            return tasksAndActionsToProcess;
        }

        private void addTask(final Task task) {
            final TaskId taskId = task.id();

            Task existingTask = pausedTasks.get(taskId);
            if (existingTask != null) {
                throw new IllegalStateException(
                    (existingTask.isActive() ? "Active" : "Standby") + " task " + taskId + " already exist in paused tasks, " +
                        "should not try to add another " + (task.isActive() ? "active" : "standby") + " task with the same id. "
                        + BUG_ERROR_MESSAGE);
            }
            existingTask = updatingTasks.get(taskId);
            if (existingTask != null) {
                throw new IllegalStateException(
                    (existingTask.isActive() ? "Active" : "Standby") + " task " + taskId + " already exist in updating tasks, " +
                        "should not try to add another " + (task.isActive() ? "active" : "standby") + " task with the same id. "
                        + BUG_ERROR_MESSAGE);
            }

            if (isStateless(task)) {
                addToRestoredTasks((StreamTask) task);
                log.info("Stateless active task " + taskId + " was added to the restored tasks of the state updater");
            } else if (topologyMetadata.isPaused(taskId.topologyName())) {
                pausedTasks.put(taskId, task);
                changelogReader.register(task.changelogPartitions(), task.stateManager());
                log.debug((task.isActive() ? "Active" : "Standby")
                    + " task " + taskId + " was directly added to the paused tasks.");
            } else {
                updatingTasks.put(taskId, task);
                changelogReader.register(task.changelogPartitions(), task.stateManager());
                if (task.isActive()) {
                    log.info("Stateful active task " + taskId + " was added to the state updater");
                    changelogReader.enforceRestoreActive();
                } else {
                    log.info("Standby task " + taskId + " was added to the state updater");
                    if (updatingTasks.size() == 1) {
                        changelogReader.transitToUpdateStandby();
                    }
                }
            }
        }

        private void removeTask(final TaskId taskId) {
            final Task task;
            if (updatingTasks.containsKey(taskId)) {
                task = updatingTasks.get(taskId);
                task.maybeCheckpoint(true);
                final Collection<TopicPartition> changelogPartitions = task.changelogPartitions();
                changelogReader.unregister(changelogPartitions);