void markChangelogAsCorrupted(final Collection<TopicPartition> partitions) {
        final Collection<TopicPartition> partitionsToMarkAsCorrupted = new LinkedList<>(partitions);
        for (final StateStoreMetadata storeMetadata : stores.values()) {
            if (partitionsToMarkAsCorrupted.contains(storeMetadata.changelogPartition)) {
                storeMetadata.corrupted = true;
                partitionsToMarkAsCorrupted.remove(storeMetadata.changelogPartition);
            }
        }

        if (!partitionsToMarkAsCorrupted.isEmpty()) {
            throw new IllegalStateException("Some partitions " + partitionsToMarkAsCorrupted + " are not contained in " +
                "the store list of task " + taskId + " marking as corrupted, this is not expected");
        }
    }
@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