private Set<Task> getTasksFromPartitions(final Map<TaskId, Task> tasks,
                                             final Set<TopicPartition> partitions) {
        final Set<Task> result = new HashSet<>();
        for (final TopicPartition partition : partitions) {
            result.add(tasks.get(changelogs.get(partition).stateManager.taskId()));
        }
        return result;
    }
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