@Override
    public long restore(final Map<TaskId, Task> tasks) {
        initializeChangelogs(tasks, registeredChangelogs());

        if (!activeRestoringChangelogs().isEmpty() && state == ChangelogReaderState.STANDBY_UPDATING) {
            throw new IllegalStateException("Should not be in standby updating state if there are still un-completed active changelogs");
        }

        long totalRestored = 0L;
        if (allChangelogsCompleted()) {
            log.debug("Finished restoring all changelogs {}", changelogs.keySet());
            return totalRestored;
        }

        final Set<TopicPartition> restoringChangelogs = restoringChangelogs();
        if (!restoringChangelogs.isEmpty()) {
            final ConsumerRecords<byte[], byte[]> polledRecords = pollRecordsFromRestoreConsumer(tasks, restoringChangelogs);

            for (final TopicPartition partition : polledRecords.partitions()) {
                bufferChangelogRecords(restoringChangelogByPartition(partition), polledRecords.records(partition));
            }

            for (final TopicPartition partition : restoringChangelogs) {
                // even if some partition do not have any accumulated data, we still trigger
                // restoring since some changelog may not need to restore any at all, and the
                // restore to end check needs to be executed still.
                // TODO: we always try to restore as a batch when some records are accumulated, which may result in
                //       small batches; this can be optimized in the future, e.g. wait longer for larger batches.
                final TaskId taskId = changelogs.get(partition).stateManager.taskId();
                final Task task = tasks.get(taskId);

                try {
                    final ChangelogMetadata changelogMetadata = changelogs.get(partition);
                    totalRestored += restoreChangelog(task, changelogMetadata);
                } catch (final TimeoutException timeoutException) {
                    tasks.get(taskId).maybeInitTaskTimeoutOrThrow(
                        time.milliseconds(),
                        timeoutException
                    );
                }
            }

            maybeUpdateLimitOffsetsForStandbyChangelogs(tasks);

            maybeLogRestorationProgress();
        }

        return totalRestored;
    }
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