class AssignedStreamsTasks extends AssignedTasks<StreamTask> implements RestoringTasks {

    int process(final long now) {
        int processed = 0;

        for (final StreamTask task : running.values()) {
            try {
                if (task.isProcessable(now) && task.process()) {
                    processed++;
                }
            } catch (final TaskMigratedException e) {
                log.info("Failed to process stream task {} since it got migrated to another thread already. " +
                        "Will trigger a new rebalance and close all tasks as zombies together.", task.id());
                throw e;
            } catch (final RuntimeException e) {
                log.error("Failed to process stream task {} due to the following error:", task.id(), e);
                throw e;
            }
        }

        return processed;
    }

      AssignedStreamsTasks(final LogContext logContext);

}