private void removeAllProducedSensors() {
        for (final Sensor sensor : producedSensorByTopic.values()) {
            streamsMetrics.removeSensor(sensor);
        }
    }
@throws StreamsException fatal error that should cause the thread to die
@throws TaskMigratedException recoverable error that would cause the task to be removed
streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java