streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
void handleRevocation(final Collection<TopicPartition> revokedPartitions) {
        final Set<TopicPartition> remainingPartitions = new HashSet<>(revokedPartitions);

        final Map<TaskId, Map<TopicPartition, OffsetAndMetadata>> consumedOffsetsAndMetadataPerTask = new HashMap<>();
        final Set<Task> suspended = new HashSet<>();
        for (final Task task : tasks.values()) {
            if (remainingPartitions.containsAll(task.inputPartitions())) {
                task.prepareSuspend();
                final Map<TopicPartition, OffsetAndMetadata> committableOffsets = task.committableOffsetsAndMetadata();
                if (!committableOffsets.isEmpty()) {
                    consumedOffsetsAndMetadataPerTask.put(task.id(), committableOffsets);
                }
                suspended.add(task);
            } else {
                if (task.isActive() && task.commitNeeded()) {
                    task.prepareCommit();
                    final Map<TopicPartition, OffsetAndMetadata> committableOffsets = task.committableOffsetsAndMetadata();
                    if (!committableOffsets.isEmpty()) {
                        consumedOffsetsAndMetadataPerTask.put(task.id(), committableOffsets);
                    }
                }
            }
            remainingPartitions.removeAll(task.inputPartitions());
        }

        if (!consumedOffsetsAndMetadataPerTask.isEmpty()) {
            commitOffsetsOrTransaction(consumedOffsetsAndMetadataPerTask);
        }

        for (final Task task : tasks.values()) {
            if (consumedOffsetsAndMetadataPerTask.containsKey(task.id())) {
                if (suspended.contains(task)) {
                    task.suspend();
                } else {
                    task.postCommit();
                }
            }
        }

        if (!remainingPartitions.isEmpty()) {
            log.warn("The following partitions {} are missing from the task partitions. It could potentially " +
                         "due to race condition of consumer detecting the heartbeat failure, or the tasks " +
                         "have been cleaned up by the handleAssignment callback.", remainingPartitions);
        }
    }
/**
     * @throws TaskMigratedException if the task producer got fenced (EOS only)
     */
@throws TaskMigratedException if the task producer got fenced (EOS only)
@throws TaskMigratedException if the task producer got fenced ( EOS only )
void handleRevocation(final Collection<TopicPartition> revokedPartitions) {
        final Set<TopicPartition> remainingPartitions = new HashSet<>(revokedPartitions);
        final Map<TaskId, Map<TopicPartition, OffsetAndMetadata>> consumedOffsetsAndMetadataPerTask = new HashMap<>();
        final Set<Task> suspended = new HashSet<>();
        for (final Task task : tasks.values()) {
            if (remainingPartitions.containsAll(task.inputPartitions())) {
                task.prepareSuspend();
                final Map<TopicPartition, OffsetAndMetadata> committableOffsets = task.committableOffsetsAndMetadata();
                if (!committableOffsets.isEmpty()) {
                    consumedOffsetsAndMetadataPerTask.put(task.id(), committableOffsets);
                }
                suspended.add(task);
            } else {
                if (task.isActive() && task.commitNeeded()) {
                    task.prepareCommit();
                    final Map<TopicPartition, OffsetAndMetadata> committableOffsets = task.committableOffsetsAndMetadata();
                    if (!committableOffsets.isEmpty()) {
                        consumedOffsetsAndMetadataPerTask.put(task.id(), committableOffsets);
                    }
                }
            }
            remainingPartitions.removeAll(task.inputPartitions());
        }
        if (!consumedOffsetsAndMetadataPerTask.isEmpty()) {
            commitOffsetsOrTransaction(consumedOffsetsAndMetadataPerTask);
        }
        for (final Task task : tasks.values()) {
            if (consumedOffsetsAndMetadataPerTask.containsKey(task.id())) {
                if (suspended.contains(task)) {
                    task.suspend();
                } else {
                    task.postCommit();
                }
            }
        }
        if (!remainingPartitions.isEmpty()) {
            log.warn("The following partitions {} are missing from the task partitions. It could potentially " +
                         "due to race condition of consumer detecting the heartbeat failure, or the tasks " +
                         "have been cleaned up by the handleAssignment callback.", remainingPartitions);
        }
    }
void handleRevocation ( final Collection < TopicPartition > revokedPartitions ) { final Set < TopicPartition > remainingPartitions = new HashSet < > ( revokedPartitions ) ; final Map < TaskId , Map < TopicPartition , OffsetAndMetadata > > consumedOffsetsAndMetadataPerTask = new HashMap < > ( ) ; final Set < Task > suspended = new HashSet < > ( ) ; for ( final Task task : tasks . values ( ) ) { if ( remainingPartitions . containsAll ( task . inputPartitions ( ) ) ) { task . prepareSuspend ( ) ; final Map < TopicPartition , OffsetAndMetadata > committableOffsets = task . committableOffsetsAndMetadata ( ) ; if ( ! committableOffsets . isEmpty ( ) ) { consumedOffsetsAndMetadataPerTask . put ( task . id ( ) , committableOffsets ) ; } suspended . add ( task ) ; } else { if ( task . isActive ( ) && task . commitNeeded ( ) ) { task . prepareCommit ( ) ; final Map < TopicPartition , OffsetAndMetadata > committableOffsets = task . committableOffsetsAndMetadata ( ) ; if ( ! committableOffsets . isEmpty ( ) ) { consumedOffsetsAndMetadataPerTask . put ( task . id ( ) , committableOffsets ) ; } } } remainingPartitions . removeAll ( task . inputPartitions ( ) ) ; } if ( ! consumedOffsetsAndMetadataPerTask . isEmpty ( ) ) { commitOffsetsOrTransaction ( consumedOffsetsAndMetadataPerTask ) ; } for ( final Task task : tasks . values ( ) ) { if ( consumedOffsetsAndMetadataPerTask . containsKey ( task . id ( ) ) ) { if ( suspended . contains ( task ) ) { task . suspend ( ) ; } else { task . postCommit ( ) ; } } } if ( ! remainingPartitions . isEmpty ( ) ) { log . warn ( "The following partitions {} are missing from the task partitions. It could potentially " + "due to race condition of consumer detecting the heartbeat failure, or the tasks " + "have been cleaned up by the handleAssignment callback." , remainingPartitions ) ; } }
@throws TaskMigratedException if the task producer got fenced (EOS only)
@throws TaskMigratedException if the task producer got fenced ( EOS only )