private void maybeInitTaskTimeoutOrThrow(final Set<Task> tasks,
                                             final Exception cause) {
        final long now = time.milliseconds();
        tasks.forEach(t -> t.maybeInitTaskTimeoutOrThrow(now, cause));
    }
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