private void clearTaskTimeout(final Set<Task> tasks) {
        tasks.forEach(t -> {
            if (t != null) {
                t.clearTaskTimeout();
            }
        });
    }
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