code: public static Collection<WorkUnitState> mergeAllSplitWorkUnits(FileSystem fs, Collection<WorkUnitState> workUnits)
        throws IOException {
      ListMultimap<CopyableFile, WorkUnitState> splitWorkUnitsMap = ArrayListMultimap.create();
      for (WorkUnitState workUnit : workUnits) {
        if (isSplitWorkUnit(workUnit)) {
          CopyableFile copyableFile = (CopyableFile) CopySource.deserializeCopyEntity(workUnit);
          splitWorkUnitsMap.put(copyableFile, workUnit);
        }
      }
  
      for (CopyableFile file : splitWorkUnitsMap.keySet()) {
        log.info(String.format("Merging split file %s.", file.getDestination()));
  
        WorkUnitState oldWorkUnit = splitWorkUnitsMap.get(file).get(0);
        Path outputDir = FileAwareInputStreamDataWriter.getOutputDir(oldWorkUnit);
        CopyEntity.DatasetAndPartition datasetAndPartition =
            file.getDatasetAndPartition(CopySource.deserializeCopyableDataset(oldWorkUnit));
        Path parentPath = FileAwareInputStreamDataWriter.getOutputFilePath(file, outputDir, datasetAndPartition)
            .getParent();
  
        WorkUnitState newWorkUnit = mergeSplits(fs, file, splitWorkUnitsMap.get(file), parentPath);
  
        for (WorkUnitState wu : splitWorkUnitsMap.get(file)) {
          // Set to committed so that task states will not fail
          wu.setWorkingState(WorkUnitState.WorkingState.COMMITTED);
          workUnits.remove(wu);
        }
        workUnits.add(newWorkUnit);
      }
      return workUnits;
    }
docs: Finds all split work units in the input collection and merges the file parts into the expected output files.
  @param fs {@link FileSystem} where file parts exist.
  @param workUnits Collection of {@link WorkUnitState}s possibly containing split work units.
  @return The collection of {@link WorkUnitState}s where split work units for each file have been merged.
  @throws IOException