private void prepareChangelogs(final Map<TaskId, Task> tasks,
                                   final Set<ChangelogMetadata> newPartitionsToRestore) {
        // separate those who do not have the current offset loaded from checkpoint
        final Set<TopicPartition> newPartitionsWithoutStartOffset = new HashSet<>();

        for (final ChangelogMetadata changelogMetadata : newPartitionsToRestore) {
            final StateStoreMetadata storeMetadata = changelogMetadata.storeMetadata;
            final TopicPartition partition = storeMetadata.changelogPartition();
            final Long currentOffset = storeMetadata.offset();
            final Long endOffset = changelogs.get(partition).restoreEndOffset;

            if (currentOffset != null) {
                // the current offset is the offset of the last record, so we should set the position
                // as that offset + 1 as the "next" record to fetch; seek is not a blocking call so
                // there's nothing to capture
                restoreConsumer.seek(partition, currentOffset + 1);

                log.debug("Start restoring changelog partition {} from current offset {} to end offset {}.",
                    partition, currentOffset, recordEndOffset(endOffset));
            } else {
                log.debug("Start restoring changelog partition {} from the beginning offset to end offset {} " +
                    "since we cannot find current offset.", partition, recordEndOffset(endOffset));

                newPartitionsWithoutStartOffset.add(partition);
            }
        }

        // optimization: batch all seek-to-beginning offsets in a single request
        //               seek is not a blocking call so there's nothing to capture
        if (!newPartitionsWithoutStartOffset.isEmpty()) {
            restoreConsumer.seekToBeginning(newPartitionsWithoutStartOffset);
        }

        for (final ChangelogMetadata changelogMetadata : newPartitionsToRestore) {
            final StateStoreMetadata storeMetadata = changelogMetadata.storeMetadata;
            final TopicPartition partition = storeMetadata.changelogPartition();
            final String storeName = storeMetadata.store().name();
            long startOffset = 0L;
            try {
                startOffset = restoreConsumer.position(partition);
            } catch (final TimeoutException swallow) {
                // if we cannot find the starting position at the beginning, just use the default 0L
            } catch (final KafkaException e) {
                // this also includes InvalidOffsetException, which should not happen under normal
                // execution, hence it is also okay to wrap it as fatal StreamsException
                throw new StreamsException("Restore consumer get unexpected error trying to get the position " +
                        " of " + partition, e);
            }
            if (changelogMetadata.stateManager.taskType() == Task.TaskType.ACTIVE) {
                try {
                    stateRestoreListener.onRestoreStart(partition, storeName, startOffset, changelogMetadata.restoreEndOffset);
                } catch (final Exception e) {
                    throw new StreamsException("State restore listener failed on restore start", e);
                }

                final TaskId taskId = changelogMetadata.stateManager.taskId();
                final Task task = tasks.get(taskId);
                // if the log is truncated between when we get the log end offset and when we get the
                // consumer position, then it's possible that the difference become negative and there's actually
                // no records to restore; in this case we just initialize the sensor to zero
                final long recordsToRestore = Math.max(changelogMetadata.restoreEndOffset - startOffset, 0L);
                task.recordRestoration(time, recordsToRestore, true);
            }  else if (changelogMetadata.stateManager.taskType() == TaskType.STANDBY) {
                try {
                    standbyUpdateListener.onUpdateStart(partition, storeName, startOffset);
                } catch (final Exception e) {
                    throw new StreamsException("Standby updater listener failed on update start");
                }
            }
        }
    }
restore a changelog with its buffered records if there's any; for active changelogs also check if
it has completed the restoration and can transit to COMPLETED state and trigger restore callbacks

@return number of records restored
streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java