private StateStoreMetadata findStore(final TopicPartition changelogPartition) {
        final List<StateStoreMetadata> found = stores.values().stream()
            .filter(metadata -> changelogPartition.equals(metadata.changelogPartition))
            .collect(Collectors.toList());

        if (found.size() > 1) {
            throw new IllegalStateException("Multiple state stores are found for changelog partition " + changelogPartition +
                ", this should never happen: " + found);
        }

        return found.isEmpty() ? null : found.get(0);
    }
Alternative to {@link #close()} that just resets the changelogs without closing any of the underlying state
or unregistering the stores themselves
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java