boolean changelogAsSource(final TopicPartition partition) {
        return sourcePartitions.contains(partition);
    }
@throws ProcessorStateException if the task directory does not exist and could not be created
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java