@Override
    public void register(final Set<TopicPartition> changelogPartitions, final ProcessorStateManager stateManager) {
        for (final TopicPartition changelogPartition : changelogPartitions) {
            register(changelogPartition, stateManager);
        }
    }
Since it is shared for multiple tasks and hence multiple state managers, the registration would take its
corresponding state manager as well for restoring.
streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java