Pack the list of {@code WorkUnit}s into {@code MultiWorkUnit}s.

TODO: this is currently a simple round-robin packing. More sophisticated bin packing may be necessary
if the round-robin approach leads to mapper skew.
426511
private static List<WorkUnit> pack(List<WorkUnit> workUnits, int numOfMultiWorkunits) {
    Preconditions.checkArgument(numOfMultiWorkunits > 0);

    if (workUnits.size() <= numOfMultiWorkunits) {
      return workUnits;
    }
    List<WorkUnit> result = Lists.newArrayListWithCapacity(numOfMultiWorkunits);
    for (int i = 0; i < numOfMultiWorkunits; i++) {
      result.add(MultiWorkUnit.createEmpty());
    }
    for (int i = 0; i < workUnits.size(); i++) {
      ((MultiWorkUnit) result.get(i % numOfMultiWorkunits)).addWorkUnit(workUnits.get(i));
    }
    return result;
  }
518097
@Override
  public PackingPlan repack(PackingPlan currentPackingPlan, Map<String, Integer> componentChanges) {
    this.numContainers = currentPackingPlan.getContainers().size();
    resetToFirstContainer();

    int additionalContainers = computeNumAdditionalContainers(componentChanges, currentPackingPlan);
    if (additionalContainers > 0) {
      increaseNumContainers(additionalContainers);
      LOG.info(String.format(
          "Allocated %s additional containers for repack bring the number of containers to %s.",
          additionalContainers, this.numContainers));
    }

    while (true) {
      try {
        PackingPlanBuilder planBuilder = newPackingPlanBuilder(currentPackingPlan);
        planBuilder.updateNumContainers(numContainers);
        planBuilder = getResourceCompliantRRAllocation(planBuilder, componentChanges);

        return planBuilder.build();

      } catch (ConstraintViolationException e) {
        //Not enough containers. Adjust the number of containers.
        LOG.info(String.format(
            "%s Increasing the number of containers to %s and attempting to repack again.",
            e.getMessage(), this.numContainers + 1));
        retryWithAdditionalContainer();
      }
    }
  }
778818
private List<WorkUnit> materializeWorkUnitList(WorkUnitStream workUnitStream) {
    if (!workUnitStream.isFiniteStream()) {
      throw new UnsupportedOperationException("Cannot materialize an infinite work unit stream.");
    }
    return Lists.newArrayList(workUnitStream.getWorkUnits());
  }
863106
public static void unpack(InputStream is, File outputDir, NameMapper mapper) {
    unpack(is, outputDir, mapper, null);
  }
510781
private List<Row> createWorkPatternAssignmentRowList(String workPatterns) throws ParseException
   {
      List<Row> list = new ArrayList<Row>();
      String[] patterns = workPatterns.split(",|:");
      int index = 1;
      while (index < patterns.length)
      {
         Integer workPattern = Integer.valueOf(patterns[index + 1]);
         Date startDate = DatatypeConverter.parseBasicTimestamp(patterns[index + 3]);
         Date endDate = DatatypeConverter.parseBasicTimestamp(patterns[index + 4]);

         Map<String, Object> map = new HashMap<String, Object>();
         map.put("WORK_PATTERN", workPattern);
         map.put("START_DATE", startDate);
         map.put("END_DATE", endDate);

         list.add(new MapRow(map));

         index += 5;
      }

      return list;
   }
1066997
void scheduleHeronWorkers(PackingPlan topologyPacking) throws ContainerAllocationException {
    this.componentRamMap = topologyPacking.getComponentRamDistribution();
    scheduleHeronWorkers(topologyPacking.getContainers());
  }
567719
private void prepareHadoopJob(List<WorkUnit> workUnits) throws IOException {
    TimingEvent mrJobSetupTimer = this.eventSubmitter.getTimingEvent(TimingEvent.RunJobTimings.MR_JOB_SETUP);

    // Add dependent jars/files
    addDependencies(this.job.getConfiguration());

    this.job.setJarByClass(MRJobLauncher.class);
    this.job.setMapperClass(TaskRunner.class);

    // The job is mapper-only
    this.job.setNumReduceTasks(0);

    this.job.setInputFormatClass(GobblinWorkUnitsInputFormat.class);
    this.job.setOutputFormatClass(GobblinOutputFormat.class);
    this.job.setMapOutputKeyClass(NullWritable.class);
    this.job.setMapOutputValueClass(NullWritable.class);

    // Set speculative execution

    this.job.setSpeculativeExecution(isSpeculativeExecutionEnabled(this.jobProps));

    this.job.getConfiguration().set("mapreduce.job.user.classpath.first", "true");

    // Job input path is where input work unit files are stored

    // Prepare job input
    prepareJobInput(workUnits);
    FileInputFormat.addInputPath(this.job, this.jobInputPath);

    // Job output path is where serialized task states are stored
    FileOutputFormat.setOutputPath(this.job, this.jobOutputPath);

    // Serialize source state to a file which will be picked up by the mappers
    serializeJobState(this.fs, this.mrJobDir, this.conf, this.jobContext.getJobState(), this.job);

    if (this.jobProps.containsKey(ConfigurationKeys.MR_JOB_MAX_MAPPERS_KEY)) {
      GobblinWorkUnitsInputFormat.setMaxMappers(this.job,
          Integer.parseInt(this.jobProps.getProperty(ConfigurationKeys.MR_JOB_MAX_MAPPERS_KEY)));
    }

    this.job.getConfiguration().set(GOBBLIN_JOB_INTERRUPT_PATH_KEY, this.interruptPath.toString());

    mrJobSetupTimer.stop();
  }
1200862
public Boolean setPackingPlan(PackingPlans.PackingPlan packingPlan, String topologyName) {
    return awaitResult(delegate.setPackingPlan(packingPlan, topologyName));
  }
699489
protected void createWorkUnits(SourceState state)
      throws IOException {
    createWorkunitsFromPreviousState(state);
    if (this.datasets.isEmpty()) {
      return;
    }
    for (HivePartitionDataset dataset : this.datasets) {
      Optional<String> owner = dataset.getOwner();
      if (workUnitsExceeded()) {
        log.info("Workunits exceeded");
        setJobWatermark(state, dataset.datasetURN());
        return;
      }
      if (!this.policy.shouldPurge(dataset)) {
        continue;
      }
      WorkUnit workUnit = createNewWorkUnit(dataset);
      log.info("Created new work unit with partition " + workUnit.getProp(ComplianceConfigurationKeys.PARTITION_NAME));
      this.workUnitMap.put(workUnit.getProp(ComplianceConfigurationKeys.PARTITION_NAME), workUnit);
      this.workUnitsCreatedCount++;
    }
    if (!state.contains(ComplianceConfigurationKeys.HIVE_PURGER_WATERMARK)) {
      this.setJobWatermark(state, ComplianceConfigurationKeys.NO_PREVIOUS_WATERMARK);
    }
  }
1300614
private Map<Integer, List<Row>> createWorkPatternAssignmentMap(List<Row> rows) throws ParseException
   {
      Map<Integer, List<Row>> map = new HashMap<Integer, List<Row>>();
      for (Row row : rows)
      {
         Integer calendarID = row.getInteger("ID");
         String workPatterns = row.getString("WORK_PATTERNS");
         map.put(calendarID, createWorkPatternAssignmentRowList(workPatterns));
      }
      return map;
   }
1370643
public static List<ItemStack> consolidateItemStacks(List<ItemStack> inputStacks)
    {
        // Horrible n^2 method - we should do something nicer if this ever becomes a bottleneck.
        List<ItemStack> outputStacks = new ArrayList<ItemStack>();
        for (ItemStack sourceIS : inputStacks)
        {
            boolean bFound = false;
            for (ItemStack destIS : outputStacks)
            {
                if (destIS != null && sourceIS != null && itemStackIngredientsMatch(destIS, sourceIS))
                {
                    bFound = true;
                    destIS.setCount(destIS.getCount() + sourceIS.getCount());
                }
            }
            if (!bFound)
                outputStacks.add(sourceIS.copy());
        }
        return outputStacks;
    }
984185
public List<WorkUnit> getWorkUnits() {
    ImmutableList.Builder<WorkUnit> allWorkUnits = ImmutableList.builder();
    for (List<WorkUnit> workUnits : this.workUnitsMap.values()) {
      allWorkUnits.addAll(workUnits);
    }
    return allWorkUnits.build();
  }
494550
public void setWorkingWeeks(final Map<String, WorkingWeek> workingWeeks) {
        final Map<String, WorkingWeek> ww = new HashMap<>();
        ww.putAll(workingWeeks);
        this.workingWeeks = ww;
    }
426015
public void unionWith(Multiset<O> multiset) throws ParameterException {
		Validate.notNull(multiset);
		for (O o : multiset.support()) {
			setMultiplicity(o, Math.max(multiplicity(o), multiset.multiplicity(o)));
		}
	}
1018960
public WorkUnit createWorkUnit(SubTask execUnit) {
        WorkUnit wu = new WorkUnit(this, execUnit);
        workUnits.add(wu);
        return wu;
    }
117805
public PackingPlan cloneWithHomogeneousScheduledResource() {
    Resource maxResource = getMaxContainerResources();
    Set<ContainerPlan> updatedContainers = new LinkedHashSet<>();
    for (ContainerPlan container : getContainers()) {
      updatedContainers.add(container.cloneWithScheduledResource(maxResource));
    }

    return new PackingPlan(getId(), updatedContainers);
  }
1162411
private String persistWorkUnit(final Path workUnitFileDir, final WorkUnit workUnit, ParallelRunner stateSerDeRunner)
      throws IOException {
    final StateStore stateStore;
    String workUnitFileName = workUnit.getId();

    if (workUnit instanceof MultiWorkUnit) {
      workUnitFileName += MULTI_WORK_UNIT_FILE_EXTENSION;
      stateStore = stateStores.getMwuStateStore();
    } else {
      workUnitFileName += WORK_UNIT_FILE_EXTENSION;
      stateStore = stateStores.getWuStateStore();
    }

    Path workUnitFile = new Path(workUnitFileDir, workUnitFileName);
    final String fileName = workUnitFile.getName();
    final String storeName = workUnitFile.getParent().getName();
    stateSerDeRunner.submitCallable(new Callable<Void>() {
      @Override
      public Void call() throws Exception {
        stateStore.put(storeName, fileName, workUnit);
        return null;
      }
    }, "Serialize state to store " + storeName + " file " + fileName);

    return workUnitFile.toString();
  }