code: public void commit()
        throws IOException {
      if (this.tasks == null || this.tasks.isEmpty()) {
        log.warn("No tasks to be committed in container " + containerIdOptional.or(""));
        return;
      }
      Iterator<Callable<Void>> callableIterator =
          Iterators.transform(this.tasks.iterator(), new Function<Task, Callable<Void>>() {
            @Override
            public Callable<Void> apply(final Task task) {
              return new Callable<Void>() {
                @Nullable
                @Override
                public Void call()
                    throws Exception {
                  task.commit();
                  return null;
                }
              };
            }
          });
  
      try {
        List<Either<Void, ExecutionException>> executionResults =
            new IteratorExecutor<>(callableIterator, this.getTaskCommitThreadPoolSize(),
                ExecutorsUtils.newDaemonThreadFactory(Optional.of(log), Optional.of("Task-committing-pool-%d")))
                .executeAndGetResults();
        IteratorExecutor.logFailures(executionResults, log, 10);
      } catch (InterruptedException ie) {
        log.error("Committing of tasks interrupted. Aborting.");
        throw new RuntimeException(ie);
      } finally {
        persistTaskStateStore();
        if (this.cleanupCommitSteps != null) {
          for (CommitStep cleanupCommitStep : this.cleanupCommitSteps) {
            log.info("Executing additional commit step.");
            cleanupCommitStep.execute();
          }
        }
      }
    }
docs: Commit {@link #tasks} by 1. calling {@link Task#commit()} in parallel; 2. executing any additional {@link CommitStep};
  3. persist task statestore.
  @throws IOException