private void addChangelogsToRestoreConsumer(final Set<TopicPartition> partitions) {
        if (partitions.isEmpty())
            return;

        final Set<TopicPartition> assignment = new HashSet<>(restoreConsumer.assignment());

        // the current assignment should not contain any of the new partitions
        if (assignment.removeAll(partitions)) {
            throw new IllegalStateException("The current assignment " + restoreConsumer.assignment() + " " +
                "already contains some of the new partitions " + partitions);
        }
        assignment.addAll(partitions);
        restoreConsumer.assign(assignment);

        log.debug("Added partitions {} to the restore consumer, current assignment is {}", partitions, assignment);
    }
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