code: public static GobblinMultiTaskAttempt runWorkUnits(String jobId, String containerId, JobState jobState,
        List<WorkUnit> workUnits, TaskStateTracker taskStateTracker, TaskExecutor taskExecutor,
        StateStore<TaskState> taskStateStore,
        CommitPolicy multiTaskAttemptCommitPolicy, SharedResourcesBroker<GobblinScopeTypes> jobBroker,
        Predicate<GobblinMultiTaskAttempt> interruptionPredicate)
        throws IOException, InterruptedException {
  
      // dump the work unit if tracking logs are enabled
      if (jobState.getPropAsBoolean(ConfigurationKeys.WORK_UNIT_ENABLE_TRACKING_LOGS)) {
        Logger log = LoggerFactory.getLogger(GobblinMultiTaskAttempt.class.getName());
        log.info("Work unit tracking log: {}", workUnits);
      }
  
      GobblinMultiTaskAttempt multiTaskAttempt =
          new GobblinMultiTaskAttempt(workUnits.iterator(), jobId, jobState, taskStateTracker, taskExecutor,
              Optional.of(containerId), Optional.of(taskStateStore), jobBroker);
      multiTaskAttempt.setInterruptionPredicate(interruptionPredicate);
  
      multiTaskAttempt.runAndOptionallyCommitTaskAttempt(multiTaskAttemptCommitPolicy);
      return multiTaskAttempt;
    }
docs: Run a given list of {@link WorkUnit}s of a job.
  
  <p>
  This method creates {@link GobblinMultiTaskAttempt} to actually run the {@link Task}s of the {@link WorkUnit}s, and optionally commit.
  </p>
  
  @param jobId the job ID
  @param workUnits the given list of {@link WorkUnit}s to submit to run
  @param taskStateTracker a {@link TaskStateTracker} for task state tracking
  @param taskExecutor a {@link TaskExecutor} for task execution
  @param taskStateStore a {@link StateStore} for storing {@link TaskState}s
  @param multiTaskAttemptCommitPolicy {@link GobblinMultiTaskAttempt.CommitPolicy} for committing {@link GobblinMultiTaskAttempt}
  @throws IOException if there's something wrong with any IO operations
  @throws InterruptedException if the task execution gets cancelled