code: private void prepareJobInput(List<WorkUnit> workUnits) throws IOException {
      Closer closer = Closer.create();
      try {
        ParallelRunner parallelRunner = closer.register(new ParallelRunner(this.parallelRunnerThreads, this.fs));
  
        int multiTaskIdSequence = 0;
        // Serialize each work unit into a file named after the task ID
        for (WorkUnit workUnit : workUnits) {
  
          String workUnitFileName;
          if (workUnit instanceof MultiWorkUnit) {
            workUnitFileName = JobLauncherUtils.newMultiTaskId(this.jobContext.getJobId(), multiTaskIdSequence++)
                + MULTI_WORK_UNIT_FILE_EXTENSION;
          } else {
            workUnitFileName = workUnit.getProp(ConfigurationKeys.TASK_ID_KEY) + WORK_UNIT_FILE_EXTENSION;
          }
          Path workUnitFile = new Path(this.jobInputPath, workUnitFileName);
          LOG.debug("Writing work unit file " + workUnitFileName);
  
          parallelRunner.serializeToFile(workUnit, workUnitFile);
  
          // Append the work unit file path to the job input file
        }
      } catch (Throwable t) {
        throw closer.rethrow(t);
      } finally {
        closer.close();
      }
    }
docs: Prepare the job input.
  @throws IOException