private Map<TopicPartition, Long> committedOffsetForChangelogs(final Map<TaskId, Task> tasks, final Set<TopicPartition> partitions) {
        if (partitions.isEmpty()) {
            return Collections.emptyMap();
        }

        try {
            // those which do not have a committed offset would default to 0
            final ListConsumerGroupOffsetsOptions options = new ListConsumerGroupOffsetsOptions()
                    .requireStable(true);
            final ListConsumerGroupOffsetsSpec spec = new ListConsumerGroupOffsetsSpec()
                    .topicPartitions(new ArrayList<>(partitions));
            final Map<TopicPartition, Long> committedOffsets = adminClient.listConsumerGroupOffsets(Collections.singletonMap(groupId, spec))
                    .partitionsToOffsetAndMetadata(groupId).get().entrySet()
                    .stream()
                    .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue() == null ? 0L : e.getValue().offset()));

            clearTaskTimeout(getTasksFromPartitions(tasks, partitions));
            return committedOffsets;
        } catch (final TimeoutException | InterruptedException | ExecutionException retriableException) {
            log.debug("Could not retrieve the committed offsets for partitions {} due to {}, will retry in the next run loop",
                partitions, retriableException.toString());
            maybeInitTaskTimeoutOrThrow(getTasksFromPartitions(tasks, partitions), retriableException);
            return Collections.emptyMap();
        } catch (final KafkaException e) {
            throw new StreamsException(String.format("Failed to retrieve committed offsets for %s", partitions), e);
        }
    }
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