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

Code before:
            for (final StreamTask activeTask : newTasks) {
                addActiveTask(activeTask);
            }
        }
    }

    @Override
    public void addStandbyTasks(final Collection<StandbyTask> newTasks) {
        if (!newTasks.isEmpty()) {
            for (final StandbyTask standbyTask : newTasks) {
                addStandbyTask(standbyTask);
            }
        }
    }

    @Override
    public void addTask(final Task task) {
        if (task.isActive()) {
            addActiveTask((StreamTask) task);
        } else {
            addStandbyTask((StandbyTask) task);
        }
    }

    @Override
    public synchronized void addActiveTask(final StreamTask task) {
        final TaskId taskId = task.id();
        checkTaskDoesNotExist(taskId);

        activeTasksPerId.put(taskId, task);
        pendingActiveTasksToCreate.remove(taskId);
        for (final TopicPartition topicPartition : task.inputPartitions()) {
            activeTasksPerPartition.put(topicPartition, task);
        }
    }

    @Override
    public synchronized void addStandbyTask(final StandbyTask task) {
        final TaskId taskId = task.id();
        checkTaskDoesNotExist(taskId);

        standbyTasksPerId.put(taskId, task);
    }

    private void checkTaskDoesNotExist(final TaskId taskId) {
        if (activeTasksPerId.containsKey(taskId) || standbyTasksPerId.containsKey(taskId)) {
            throw new IllegalStateException("Attempted to create a task that we already own: " + taskId);
        }
    }

    @Override

Review comment (refactor):
Both original error messages say `Attempted to create an active task` -- so wondering why the original method is called `addTask`? Or was the original message not fully correct to begin with? In the end, we cannot have active and standby (with task task-id) at the same time, but it goes both ways. So it would also be a valid error message to say `Attempted to create an standby task...`?

Also, by extracting `checkTaskDoesNotExist()`, we actually get a single more general error message -- given that `checkTaskDoesNotExist()` is very short, might be better to keep the code inlined, and use two different error messages.

Code after:
            for (final StreamTask activeTask : newTasks) {
                addActiveTask(activeTask);
            }
        }
    }

    @Override
    public void addStandbyTasks(final Collection<StandbyTask> newTasks) {
        if (!newTasks.isEmpty()) {
            for (final StandbyTask standbyTask : newTasks) {
                addStandbyTask(standbyTask);
            }
        }
    }

    @Override
    public void addTask(final Task task) {
        if (task.isActive()) {
            addActiveTask((StreamTask) task);
        } else {
            addStandbyTask((StandbyTask) task);
        }
    }

    @Override
    public synchronized void addActiveTask(final StreamTask task) {
        final TaskId taskId = task.id();

        if (activeTasksPerId.containsKey(taskId)) {
            throw new IllegalStateException("Attempted to create an active task that we already own: " + taskId);
        }

        if (standbyTasksPerId.containsKey(taskId)) {
            throw new IllegalStateException("Attempted to create an active task while we already own its standby: " + taskId);
        }

        activeTasksPerId.put(taskId, task);
        pendingActiveTasksToCreate.remove(taskId);
        for (final TopicPartition topicPartition : task.inputPartitions()) {
            activeTasksPerPartition.put(topicPartition, task);
        }
    }

    @Override
    public synchronized void addStandbyTask(final StandbyTask task) {
        final TaskId taskId = task.id();

        if (standbyTasksPerId.containsKey(taskId)) {
            throw new IllegalStateException("Attempted to create an standby task that we already own: " + taskId);
        }