private boolean shouldPause(final Map<TaskId, Task> tasks, final TopicPartition partition, final TaskType taskType) {
        final ProcessorStateManager manager = changelogs.get(partition).stateManager;
        final TaskId taskId = manager.taskId();
        final Task task = tasks.get(taskId);
        if (manager.taskType() == taskType) {
            return task == null;
        }
        return false;
    }
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