streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
public void handleAssignment(final Map<TaskId, Set<TopicPartition>> activeTasks,
                                 final Map<TaskId, Set<TopicPartition>> standbyTasks) {
        log.info("Handle new assignment with:\n" +
                     "\tNew active tasks: {}\n" +
                     "\tNew standby tasks: {}\n" +
                     "\tExisting active tasks: {}\n" +
                     "\tExisting standby tasks: {}",
                 activeTasks.keySet(), standbyTasks.keySet(), activeTaskIds(), standbyTaskIds());

        final Map<TaskId, Set<TopicPartition>> activeTasksToCreate = new TreeMap<>(activeTasks);
        final Map<TaskId, Set<TopicPartition>> standbyTasksToCreate = new TreeMap<>(standbyTasks);

        // first rectify all existing tasks
        final LinkedHashMap<TaskId, RuntimeException> taskCloseExceptions = new LinkedHashMap<>();

        final Map<Task, Map<TopicPartition, Long>> checkpointPerTask = new HashMap<>();
        final Map<TaskId, Map<TopicPartition, OffsetAndMetadata>> consumedOffsetsAndMetadataPerTask = new HashMap<>();
        final Set<Task> additionalTasksForCommitting = new HashSet<>();
        final Set<Task> dirtyTasks = new HashSet<>();

        for (final Task task : tasks.values()) {
            if (activeTasks.containsKey(task.id()) && task.isActive()) {
                task.resume();
                if (task.commitNeeded()) {
                    additionalTasksForCommitting.add(task);
                }
                activeTasksToCreate.remove(task.id());
            } else if (standbyTasks.containsKey(task.id()) && !task.isActive()) {
                task.resume();
                standbyTasksToCreate.remove(task.id());
            } else /* we previously owned this task, and we don't have it anymore, or it has changed active/standby state */ {
                try {
                    final Map<TopicPartition, Long> checkpoint = task.prepareCloseClean();
                    final Map<TopicPartition, OffsetAndMetadata> committableOffsets = task
                        .committableOffsetsAndMetadata();

                    checkpointPerTask.put(task, checkpoint);
                    if (!committableOffsets.isEmpty()) {
                        consumedOffsetsAndMetadataPerTask.put(task.id(), committableOffsets);
                    }
                } catch (final RuntimeException e) {
                    final String uncleanMessage = String.format(
                        "Failed to close task %s cleanly. Attempting to close remaining tasks before re-throwing:",
                        task.id());
                    log.error(uncleanMessage, e);
                    taskCloseExceptions.put(task.id(), e);
                    // We've already recorded the exception (which is the point of clean).
                    // Now, we should go ahead and complete the close because a half-closed task is no good to anyone.
                    dirtyTasks.add(task);
                }
            }
        }

        if (!consumedOffsetsAndMetadataPerTask.isEmpty()) {
            try {
                for (final Task task : additionalTasksForCommitting) {
                    task.prepareCommit();
                    final Map<TopicPartition, OffsetAndMetadata> committableOffsets = task.committableOffsetsAndMetadata();
                    if (!committableOffsets.isEmpty()) {
                        consumedOffsetsAndMetadataPerTask.put(task.id(), committableOffsets);
                    }
                }

                commitOffsetsOrTransaction(consumedOffsetsAndMetadataPerTask);

                for (final Task task : additionalTasksForCommitting) {
                    task.postCommit();
                }
            } catch (final RuntimeException e) {
                log.error("Failed to batch commit tasks, " +
                    "will close all tasks involved in this commit as dirty by the end", e);
                dirtyTasks.addAll(additionalTasksForCommitting);
                dirtyTasks.addAll(checkpointPerTask.keySet());

                checkpointPerTask.clear();
                // Just add first taskId to re-throw by the end.
                taskCloseExceptions.put(consumedOffsetsAndMetadataPerTask.keySet().iterator().next(), e);
            }
        }

        for (final Map.Entry<Task, Map<TopicPartition, Long>> taskAndCheckpoint : checkpointPerTask.entrySet()) {
            final Task task = taskAndCheckpoint.getKey();
            final Map<TopicPartition, Long> checkpoint = taskAndCheckpoint.getValue();

            try {
                completeTaskCloseClean(task, checkpoint);
                cleanUpTaskProducer(task, taskCloseExceptions);
                tasks.remove(task.id());
            } catch (final RuntimeException e) {
                final String uncleanMessage = String.format("Failed to close task %s cleanly. Attempting to close remaining tasks before re-throwing:", task.id());
                log.error(uncleanMessage, e);
                taskCloseExceptions.put(task.id(), e);
                // We've already recorded the exception (which is the point of clean).
                // Now, we should go ahead and complete the close because a half-closed task is no good to anyone.
                dirtyTasks.add(task);
            }
        }

        for (final Task task : dirtyTasks) {
            closeTaskDirty(task);
            cleanUpTaskProducer(task, taskCloseExceptions);
            tasks.remove(task.id());
        }

        if (!taskCloseExceptions.isEmpty()) {
            for (final Map.Entry<TaskId, RuntimeException> entry : taskCloseExceptions.entrySet()) {
                if (!(entry.getValue() instanceof TaskMigratedException)) {
                    if (entry.getValue() instanceof KafkaException) {
                        log.error("Hit Kafka exception while closing for first task {}", entry.getKey());
                        throw entry.getValue();
                    } else {
                        throw new RuntimeException(
                            "Unexpected failure to close " + taskCloseExceptions.size() +
                                " task(s) [" + taskCloseExceptions.keySet() + "]. " +
                                "First unexpected exception (for task " + entry.getKey() + ") follows.", entry.getValue()
                        );
                    }
                }
            }

            final Map.Entry<TaskId, RuntimeException> first = taskCloseExceptions.entrySet().iterator().next();
            // If all exceptions are task-migrated, we would just throw the first one.
            throw first.getValue();
        }

        if (!activeTasksToCreate.isEmpty()) {
            for (final Task task : activeTaskCreator.createTasks(mainConsumer, activeTasksToCreate)) {
                addNewTask(task);
            }
        }

        if (!standbyTasksToCreate.isEmpty()) {
            for (final Task task : standbyTaskCreator.createTasks(standbyTasksToCreate)) {
                addNewTask(task);
            }
        }

        builder.addSubscribedTopicsFromAssignment(
            activeTasks.values().stream().flatMap(Collection::stream).collect(Collectors.toList()),
            logPrefix
        );
    }
/**
     * @throws TaskMigratedException if the task producer got fenced (EOS only)
     * @throws StreamsException fatal error while creating / initializing the task
     *
     * public for upgrade testing only
     */
@throws TaskMigratedException if the task producer got fenced (EOS only)
@throws StreamsException fatal error while creating / initializing the task

public for upgrade testing only
@throws TaskMigratedException if the task producer got fenced ( EOS only ) @throws StreamsException fatal error while creating / initializing the task public for upgrade testing only
public void handleAssignment(final Map<TaskId, Set<TopicPartition>> activeTasks,
                                 final Map<TaskId, Set<TopicPartition>> standbyTasks) {
        log.info("Handle new assignment with:\n" +
                     "\tNew active tasks: {}\n" +
                     "\tNew standby tasks: {}\n" +
                     "\tExisting active tasks: {}\n" +
                     "\tExisting standby tasks: {}",
                 activeTasks.keySet(), standbyTasks.keySet(), activeTaskIds(), standbyTaskIds());
        final Map<TaskId, Set<TopicPartition>> activeTasksToCreate = new TreeMap<>(activeTasks);
        final Map<TaskId, Set<TopicPartition>> standbyTasksToCreate = new TreeMap<>(standbyTasks);
        final LinkedHashMap<TaskId, RuntimeException> taskCloseExceptions = new LinkedHashMap<>();
        final Map<Task, Map<TopicPartition, Long>> checkpointPerTask = new HashMap<>();
        final Map<TaskId, Map<TopicPartition, OffsetAndMetadata>> consumedOffsetsAndMetadataPerTask = new HashMap<>();
        final Set<Task> additionalTasksForCommitting = new HashSet<>();
        final Set<Task> dirtyTasks = new HashSet<>();
        for (final Task task : tasks.values()) {
            if (activeTasks.containsKey(task.id()) && task.isActive()) {
                task.resume();
                if (task.commitNeeded()) {
                    additionalTasksForCommitting.add(task);
                }
                activeTasksToCreate.remove(task.id());
            } else if (standbyTasks.containsKey(task.id()) && !task.isActive()) {
                task.resume();
                standbyTasksToCreate.remove(task.id());
            } else /* we previously owned this task, and we don't have it anymore, or it has changed active/standby state */ {
                try {
                    final Map<TopicPartition, Long> checkpoint = task.prepareCloseClean();
                    final Map<TopicPartition, OffsetAndMetadata> committableOffsets = task
                        .committableOffsetsAndMetadata();
                    checkpointPerTask.put(task, checkpoint);
                    if (!committableOffsets.isEmpty()) {
                        consumedOffsetsAndMetadataPerTask.put(task.id(), committableOffsets);
                    }
                } catch (final RuntimeException e) {
                    final String uncleanMessage = String.format(
                        "Failed to close task %s cleanly. Attempting to close remaining tasks before re-throwing:",
                        task.id());
                    log.error(uncleanMessage, e);
                    taskCloseExceptions.put(task.id(), e);
                    dirtyTasks.add(task);
                }
            }
        }
        if (!consumedOffsetsAndMetadataPerTask.isEmpty()) {
            try {
                for (final Task task : additionalTasksForCommitting) {
                    task.prepareCommit();
                    final Map<TopicPartition, OffsetAndMetadata> committableOffsets = task.committableOffsetsAndMetadata();
                    if (!committableOffsets.isEmpty()) {
                        consumedOffsetsAndMetadataPerTask.put(task.id(), committableOffsets);
                    }
                }
                commitOffsetsOrTransaction(consumedOffsetsAndMetadataPerTask);
                for (final Task task : additionalTasksForCommitting) {
                    task.postCommit();
                }
            } catch (final RuntimeException e) {
                log.error("Failed to batch commit tasks, " +
                    "will close all tasks involved in this commit as dirty by the end", e);
                dirtyTasks.addAll(additionalTasksForCommitting);
                dirtyTasks.addAll(checkpointPerTask.keySet());
                checkpointPerTask.clear();
                taskCloseExceptions.put(consumedOffsetsAndMetadataPerTask.keySet().iterator().next(), e);
            }
        }
        for (final Map.Entry<Task, Map<TopicPartition, Long>> taskAndCheckpoint : checkpointPerTask.entrySet()) {
            final Task task = taskAndCheckpoint.getKey();
            final Map<TopicPartition, Long> checkpoint = taskAndCheckpoint.getValue();
            try {
                completeTaskCloseClean(task, checkpoint);
                cleanUpTaskProducer(task, taskCloseExceptions);
                tasks.remove(task.id());
            } catch (final RuntimeException e) {
                final String uncleanMessage = String.format("Failed to close task %s cleanly. Attempting to close remaining tasks before re-throwing:", task.id());
                log.error(uncleanMessage, e);
                taskCloseExceptions.put(task.id(), e);
                dirtyTasks.add(task);
            }
        }
        for (final Task task : dirtyTasks) {
            closeTaskDirty(task);
            cleanUpTaskProducer(task, taskCloseExceptions);
            tasks.remove(task.id());
        }
        if (!taskCloseExceptions.isEmpty()) {
            for (final Map.Entry<TaskId, RuntimeException> entry : taskCloseExceptions.entrySet()) {
                if (!(entry.getValue() instanceof TaskMigratedException)) {
                    if (entry.getValue() instanceof KafkaException) {
                        log.error("Hit Kafka exception while closing for first task {}", entry.getKey());
                        throw entry.getValue();
                    } else {
                        throw new RuntimeException(
                            "Unexpected failure to close " + taskCloseExceptions.size() +
                                " task(s) [" + taskCloseExceptions.keySet() + "]. " +
                                "First unexpected exception (for task " + entry.getKey() + ") follows.", entry.getValue()
                        );
                    }
                }
            }
            final Map.Entry<TaskId, RuntimeException> first = taskCloseExceptions.entrySet().iterator().next();
            throw first.getValue();
        }
        if (!activeTasksToCreate.isEmpty()) {
            for (final Task task : activeTaskCreator.createTasks(mainConsumer, activeTasksToCreate)) {
                addNewTask(task);
            }
        }
        if (!standbyTasksToCreate.isEmpty()) {
            for (final Task task : standbyTaskCreator.createTasks(standbyTasksToCreate)) {
                addNewTask(task);
            }
        }
        builder.addSubscribedTopicsFromAssignment(
            activeTasks.values().stream().flatMap(Collection::stream).collect(Collectors.toList()),
            logPrefix
        );
    }
public void handleAssignment ( final Map < TaskId , Set < TopicPartition > > activeTasks , final Map < TaskId , Set < TopicPartition > > standbyTasks ) { log . info ( "Handle new assignment with:\n" + "\tNew active tasks: {}\n" + "\tNew standby tasks: {}\n" + "\tExisting active tasks: {}\n" + "\tExisting standby tasks: {}" , activeTasks . keySet ( ) , standbyTasks . keySet ( ) , activeTaskIds ( ) , standbyTaskIds ( ) ) ; final Map < TaskId , Set < TopicPartition > > activeTasksToCreate = new TreeMap < > ( activeTasks ) ; final Map < TaskId , Set < TopicPartition > > standbyTasksToCreate = new TreeMap < > ( standbyTasks ) ; final LinkedHashMap < TaskId , RuntimeException > taskCloseExceptions = new LinkedHashMap < > ( ) ; final Map < Task , Map < TopicPartition , Long > > checkpointPerTask = new HashMap < > ( ) ; final Map < TaskId , Map < TopicPartition , OffsetAndMetadata > > consumedOffsetsAndMetadataPerTask = new HashMap < > ( ) ; final Set < Task > additionalTasksForCommitting = new HashSet < > ( ) ; final Set < Task > dirtyTasks = new HashSet < > ( ) ; for ( final Task task : tasks . values ( ) ) { if ( activeTasks . containsKey ( task . id ( ) ) && task . isActive ( ) ) { task . resume ( ) ; if ( task . commitNeeded ( ) ) { additionalTasksForCommitting . add ( task ) ; } activeTasksToCreate . remove ( task . id ( ) ) ; } else if ( standbyTasks . containsKey ( task . id ( ) ) && ! task . isActive ( ) ) { task . resume ( ) ; standbyTasksToCreate . remove ( task . id ( ) ) ; } else /* we previously owned this task, and we don't have it anymore, or it has changed active/standby state */ { try { final Map < TopicPartition , Long > checkpoint = task . prepareCloseClean ( ) ; final Map < TopicPartition , OffsetAndMetadata > committableOffsets = task . committableOffsetsAndMetadata ( ) ; checkpointPerTask . put ( task , checkpoint ) ; if ( ! committableOffsets . isEmpty ( ) ) { consumedOffsetsAndMetadataPerTask . put ( task . id ( ) , committableOffsets ) ; } } catch ( final RuntimeException e ) { final String uncleanMessage = String . format ( "Failed to close task %s cleanly. Attempting to close remaining tasks before re-throwing:" , task . id ( ) ) ; log . error ( uncleanMessage , e ) ; taskCloseExceptions . put ( task . id ( ) , e ) ; dirtyTasks . add ( task ) ; } } } if ( ! consumedOffsetsAndMetadataPerTask . isEmpty ( ) ) { try { for ( final Task task : additionalTasksForCommitting ) { task . prepareCommit ( ) ; final Map < TopicPartition , OffsetAndMetadata > committableOffsets = task . committableOffsetsAndMetadata ( ) ; if ( ! committableOffsets . isEmpty ( ) ) { consumedOffsetsAndMetadataPerTask . put ( task . id ( ) , committableOffsets ) ; } } commitOffsetsOrTransaction ( consumedOffsetsAndMetadataPerTask ) ; for ( final Task task : additionalTasksForCommitting ) { task . postCommit ( ) ; } } catch ( final RuntimeException e ) { log . error ( "Failed to batch commit tasks, " + "will close all tasks involved in this commit as dirty by the end" , e ) ; dirtyTasks . addAll ( additionalTasksForCommitting ) ; dirtyTasks . addAll ( checkpointPerTask . keySet ( ) ) ; checkpointPerTask . clear ( ) ; taskCloseExceptions . put ( consumedOffsetsAndMetadataPerTask . keySet ( ) . iterator ( ) . next ( ) , e ) ; } } for ( final Map . Entry < Task , Map < TopicPartition , Long > > taskAndCheckpoint : checkpointPerTask . entrySet ( ) ) { final Task task = taskAndCheckpoint . getKey ( ) ; final Map < TopicPartition , Long > checkpoint = taskAndCheckpoint . getValue ( ) ; try { completeTaskCloseClean ( task , checkpoint ) ; cleanUpTaskProducer ( task , taskCloseExceptions ) ; tasks . remove ( task . id ( ) ) ; } catch ( final RuntimeException e ) { final String uncleanMessage = String . format ( "Failed to close task %s cleanly. Attempting to close remaining tasks before re-throwing:" , task . id ( ) ) ; log . error ( uncleanMessage , e ) ; taskCloseExceptions . put ( task . id ( ) , e ) ; dirtyTasks . add ( task ) ; } } for ( final Task task : dirtyTasks ) { closeTaskDirty ( task ) ; cleanUpTaskProducer ( task , taskCloseExceptions ) ; tasks . remove ( task . id ( ) ) ; } if ( ! taskCloseExceptions . isEmpty ( ) ) { for ( final Map . Entry < TaskId , RuntimeException > entry : taskCloseExceptions . entrySet ( ) ) { if ( ! ( entry . getValue ( ) instanceof TaskMigratedException ) ) { if ( entry . getValue ( ) instanceof KafkaException ) { log . error ( "Hit Kafka exception while closing for first task {}" , entry . getKey ( ) ) ; throw entry . getValue ( ) ; } else { throw new RuntimeException ( "Unexpected failure to close " + taskCloseExceptions . size ( ) + " task(s) [" + taskCloseExceptions . keySet ( ) + "]. " + "First unexpected exception (for task " + entry . getKey ( ) + ") follows." , entry . getValue ( ) ) ; } } } final Map . Entry < TaskId , RuntimeException > first = taskCloseExceptions . entrySet ( ) . iterator ( ) . next ( ) ; throw first . getValue ( ) ; } if ( ! activeTasksToCreate . isEmpty ( ) ) { for ( final Task task : activeTaskCreator . createTasks ( mainConsumer , activeTasksToCreate ) ) { addNewTask ( task ) ; } } if ( ! standbyTasksToCreate . isEmpty ( ) ) { for ( final Task task : standbyTaskCreator . createTasks ( standbyTasksToCreate ) ) { addNewTask ( task ) ; } } builder . addSubscribedTopicsFromAssignment ( activeTasks . values ( ) . stream ( ) . flatMap ( Collection :: stream ) . collect ( Collectors . toList ( ) ) , logPrefix ) ; }
@throws TaskMigratedException if the task producer got fenced (EOS only)
 @throws StreamsException fatal error while creating / initializing the task
@throws TaskMigratedException if the task producer got fenced ( EOS only ) @throws StreamsException fatal error while creating / initializing the task
// first rectify all existing tasks // We've already recorded the exception (which is the point of clean). // Now, we should go ahead and complete the close because a half-closed task is no good to anyone. // Just add first taskId to re-throw by the end. // We've already recorded the exception (which is the point of clean). // Now, we should go ahead and complete the close because a half-closed task is no good to anyone. // If all exceptions are task-migrated, we would just throw the first one.