private List<TopicPartition> getAllChangelogTopicPartitions() {
        final List<TopicPartition> allChangelogPartitions = new ArrayList<>();
        for (final StateStoreMetadata storeMetadata : stores.values()) {
            if (storeMetadata.changelogPartition != null) {
                allChangelogPartitions.add(storeMetadata.changelogPartition);
            }
        }
        return allChangelogPartitions;
    }
@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