class AssignedStreamsTasks extends AssignedTasks<StreamTask> implements RestoringTasks {

    int maybeCommitPerUserRequested() {
        int committed = 0;
        RuntimeException firstException = null;

        for (final StreamTask task : running.values()) {
            try {
                if (task.commitRequested() && task.commitNeeded()) {
                    task.commit();
                    committed++;
                    log.debug("Committed stream task {} per user request in", task.id());
                }
            } catch (final TaskMigratedException e) {
                log.info("Failed to commit 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 t) {
                log.error("Failed to commit stream task {} due to the following error:", task.id(), t);
                if (firstException == null) {
                    firstException = t;
                }
            }
        }

        if (firstException != null) {
            throw firstException;
        }

        return committed;
    }

      AssignedStreamsTasks(final LogContext logContext);

}