streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
private void cleanupTask(final Task task) {
        // 1. remove the changelog partitions from changelog reader;
        // 2. remove the input partitions from the materialized map;
        // 3. remove the task metrics from the metrics registry
        if (!task.changelogPartitions().isEmpty()) {
            changelogReader.remove(task.changelogPartitions());
        }

        for (final TopicPartition inputPartition : task.inputPartitions()) {
            partitionToTask.remove(inputPartition);
        }

        final String threadId = Thread.currentThread().getName();
        streamsMetrics.removeAllTaskLevelSensors(threadId, task.id().toString());
    }
// Note: this MUST be called *before* actually closing the task
this MUST be called *before* actually closing the task
this MUST be called * before * actually closing the task
private void cleanupTask(final Task task) {
        if (!task.changelogPartitions().isEmpty()) {
            changelogReader.remove(task.changelogPartitions());
        }
        for (final TopicPartition inputPartition : task.inputPartitions()) {
            partitionToTask.remove(inputPartition);
        }
        final String threadId = Thread.currentThread().getName();
        streamsMetrics.removeAllTaskLevelSensors(threadId, task.id().toString());
    }
private void cleanupTask ( final Task task ) { if ( ! task . changelogPartitions ( ) . isEmpty ( ) ) { changelogReader . remove ( task . changelogPartitions ( ) ) ; } for ( final TopicPartition inputPartition : task . inputPartitions ( ) ) { partitionToTask . remove ( inputPartition ) ; } final String threadId = Thread . currentThread ( ) . getName ( ) ; streamsMetrics . removeAllTaskLevelSensors ( threadId , task . id ( ) . toString ( ) ) ; }
Note: this MUST be called *before* actually closing the task
Note : this MUST be called * before * actually closing the task
// 1. remove the changelog partitions from changelog reader; // 2. remove the input partitions from the materialized map; // 3. remove the task metrics from the metrics registry