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

        try {
            // we always use read_uncommitted to get log end offset since the last committed txn may have not advanced the LSO for EOS,
            // see KAFKA-10167 for more details
            final ListOffsetsOptions options = new ListOffsetsOptions(IsolationLevel.READ_UNCOMMITTED);
            final Map<TopicPartition, OffsetSpec> offsetSpecs =
                partitions.stream().collect(Collectors.toMap(Function.identity(), tp -> OffsetSpec.latest()));
            final Map<TopicPartition, Long> logEndOffsets = adminClient.listOffsets(offsetSpecs, options)
                .all().get().entrySet()
                .stream().collect(Collectors.toMap(Map.Entry::getKey, entry -> entry.getValue().offset()));

            clearTaskTimeout(getTasksFromPartitions(tasks, partitions));
            return logEndOffsets;
        } catch (final TimeoutException | InterruptedException | ExecutionException retriableException) {
            log.debug("Could not fetch all end offsets for {} 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 end 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