private ConsumerRecords<byte[], byte[]> pollRecordsFromRestoreConsumer(final Map<TaskId, Task> tasks,
                                                                           final Set<TopicPartition> restoringChangelogs) {
        // If we are updating only standby tasks, and are not using a separate thread, we should
        // use a non-blocking poll to unblock the processing as soon as possible.
        final boolean useNonBlockingPoll = state == ChangelogReaderState.STANDBY_UPDATING && !stateUpdaterEnabled;
        final ConsumerRecords<byte[], byte[]> polledRecords;

        try {
            pauseResumePartitions(tasks, restoringChangelogs);

            polledRecords = restoreConsumer.poll(useNonBlockingPoll ? Duration.ZERO : pollTime);

            // TODO (?) If we cannot fetch records during restore, should we trigger `task.timeout.ms` ?
            // TODO (?) If we cannot fetch records for standby task, should we trigger `task.timeout.ms` ?
        } catch (final InvalidOffsetException e) {
            log.warn("Encountered " + e.getClass().getName() +
                " fetching records from restore consumer for partitions " + e.partitions() + ", it is likely that " +
                "the consumer's position has fallen out of the topic partition offset range because the topic was " +
                "truncated or compacted on the broker, marking the corresponding tasks as corrupted and re-initializing " +
                "it later.", e);

            final Set<TaskId> corruptedTasks = new HashSet<>();
            e.partitions().forEach(partition -> corruptedTasks.add(changelogs.get(partition).stateManager.taskId()));
            throw new TaskCorruptedException(corruptedTasks, e);
        } catch (final InterruptException interruptException) {
            throw interruptException;
        } catch (final KafkaException e) {
            throw new StreamsException("Restore consumer get unexpected error polling records.", e);
        }

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