private void updatePartitionsByType(final Map<TaskId, Task> tasks,
                                        final Set<TopicPartition> restoringChangelogs,
                                        final TaskType taskType) {
        final Collection<TopicPartition> toResume =
            restoringChangelogs.stream().filter(t -> shouldResume(tasks, t, taskType)).collect(Collectors.toList());
        final Collection<TopicPartition> toPause =
            restoringChangelogs.stream().filter(t -> shouldPause(tasks, t, taskType)).collect(Collectors.toList());
        restoreConsumer.resume(toResume);
        restoreConsumer.pause(toPause);
    }
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