@Override
  public MapTask apply(MapTask input) {
    for (ParallelInstruction instruction : Apiary.listOrEmpty(input.getInstructions())) {
      ParDoInstruction parDoInstruction = instruction.getParDo();
      if (parDoInstruction != null) {
        int numOutputs = Apiary.intOrZero(parDoInstruction.getNumOutputs());
        List<MultiOutputInfo> multiOutputInfos =
            Apiary.listOrEmpty(parDoInstruction.getMultiOutputInfos());
        if (numOutputs != Apiary.listOrEmpty(instruction.getParDo().getMultiOutputInfos()).size()) {
          if (numOutputs == 1) {
            parDoInstruction.setMultiOutputInfos(
                ImmutableList.of(new MultiOutputInfo().setTag(idGenerator.getId())));
          } else {
            throw new IllegalArgumentException(
                String.format(
                    "Invalid ParDoInstruction %s, %d outputs specified, found %s tags.",
                    instruction.getSystemName(), numOutputs, multiOutputInfos));
          }
        }
      }
    }
    return input;
  }
@Test
  public void testExistingMultiOutputInfosAreUnmodified() {
    FixMultiOutputInfosOnParDoInstructions function =
        new FixMultiOutputInfosOnParDoInstructions(IdGenerators.decrementingLongs());
    MapTask output = function.apply(createMapTaskWithParDo(2, "5", "6"));
    assertEquals(createMapTaskWithParDo(2, "5", "6"), output);
  }