class AssignedTasks {

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

        for (final T task : running.values()) {
            try {
                if (task.commitNeeded()) {
                    task.commit();
                    committed++;
                }
            } catch (final TaskMigratedException e) {
                log.info("Failed to commit {} {} since it got migrated to another thread already. " +
                        "Will trigger a new rebalance and close all tasks as zombies together.", taskTypeName, task.id());
                throw e;
            } catch (final RuntimeException t) {
                log.error("Failed to commit {} {} due to the following error:",
                        taskTypeName,
                        task.id(),
                        t);
                if (firstException == null) {
                    firstException = t;
                }
            }
        }

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

        return committed;
    }

      AssignedTasks(final LogContext logContext,
                  final String taskTypeName);

}