private MRInputLegacy getMRInput(Map<String, LogicalInput> inputs) throws Exception {
    LOG.info("The inputs are: " + inputs);

    // start the mr input and wait for ready event. number of MRInput is expected to be 1
    List<Input> li = Lists.newArrayList();
    int numMRInputs = 0;
    for (LogicalInput inp : inputs.values()) {
      if (inp instanceof MRInputLegacy) {
        numMRInputs++;
        if (numMRInputs > 1) {
          throw new IllegalArgumentException("Only one MRInput is expected");
        }
        inp.start();
        li.add(inp);
      } else {
        throw new IllegalArgumentException("Expecting only one input of type MRInputLegacy." +
            " Found type: " + inp.getClass().getCanonicalName());
      }
    }

    // typically alter table .. concatenate is run on only one partition/one table,
    // so it doesn't matter if we wait for all inputs or any input to be ready.
    processorContext.waitForAnyInputReady(li);

    final MRInputLegacy theMRInput;
    if (li.size() == 1) {
      theMRInput = (MRInputLegacy) li.get(0);
      theMRInput.init();
    } else {
      throw new IllegalArgumentException("MRInputs count is expected to be 1");
    }

    return theMRInput;
  }
@param key   key to process
@param value value to process
@return true if it is not done and can take more inputs
ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MergeFileRecordProcessor.java