class AssignedStreamsTasks extends AssignedTasks<StreamTask> implements RestoringTasks {

    void updateRestored(final Collection<TopicPartition> restored) {
        if (restored.isEmpty()) {
            return;
        }
        log.trace("Stream task changelog partitions that have completed restoring so far: {}", restored);
        restoredPartitions.addAll(restored);
        for (final Iterator<Map.Entry<TaskId, StreamTask>> it = restoring.entrySet().iterator(); it.hasNext(); ) {
            final Map.Entry<TaskId, StreamTask> entry = it.next();
            final StreamTask task = entry.getValue();
            if (restoredPartitions.containsAll(task.changelogPartitions())) {
                transitionToRunning(task);
                it.remove();
                // Note that because we add back all restored partitions at the top of this loop, clearing them from
                // restoredPartitions here doesn't really matter. We do it anyway as it is the correct thing to do,
                // and may matter with future changes.
                removeFromRestoredPartitions(task);
                removeFromRestoringByPartition(task);

                log.debug("Stream task {} completed restoration as all its changelog partitions {} have been applied to restore state",
                    task.id(),
                    task.changelogPartitions());
            } else {
                if (log.isTraceEnabled()) {
                    final HashSet<TopicPartition> outstandingPartitions = new HashSet<>(task.changelogPartitions());
                    outstandingPartitions.removeAll(restoredPartitions);
                    log.trace("Stream task {} cannot resume processing yet since some of its changelog partitions have not completed restoring: {}",
                        task.id(),
                        outstandingPartitions);
                }
            }
        }
        if (allTasksRunning()) {
            restoredPartitions.clear();

            if (!restoringByPartition.isEmpty()) {
                log.error("Finished restoring all tasks but found leftover partitions in restoringByPartition: {}",
                    restoringByPartition);
                throw new IllegalStateException("Restoration is complete but not all partitions were cleared.");
            }
        }
    }

      AssignedStreamsTasks(final LogContext logContext);

}