Set<TopicPartition> changelogPartitions() {
        return Collections.unmodifiableSet(changelogOffsets().keySet());
    }
@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