@Override
  public MutableNetwork<Node, Edge> apply(MapTask mapTask) {
    List<ParallelInstruction> parallelInstructions = Apiary.listOrEmpty(mapTask.getInstructions());
    MutableNetwork<Node, Edge> network =
        NetworkBuilder.directed()
            .allowsSelfLoops(false)
            .allowsParallelEdges(true)
            .expectedNodeCount(parallelInstructions.size() * 2)
            .build();

    // Add all the instruction nodes and output nodes
    ParallelInstructionNode[] instructionNodes =
        new ParallelInstructionNode[parallelInstructions.size()];
    InstructionOutputNode[][] outputNodes =
        new InstructionOutputNode[parallelInstructions.size()][];
    for (int i = 0; i < parallelInstructions.size(); ++i) {
      // InstructionOutputNode's are the source of truth on instruction outputs.
      // Clear the instruction's outputs to reduce chance for confusion.
      List<InstructionOutput> outputs =
          Apiary.listOrEmpty(parallelInstructions.get(i).getOutputs());
      outputNodes[i] = new InstructionOutputNode[outputs.size()];

      JsonFactory factory =
          MoreObjects.firstNonNull(mapTask.getFactory(), Transport.getJsonFactory());
      ParallelInstruction parallelInstruction =
          clone(factory, parallelInstructions.get(i)).setOutputs(null);

      ParallelInstructionNode instructionNode =
          ParallelInstructionNode.create(parallelInstruction, Nodes.ExecutionLocation.UNKNOWN);
      instructionNodes[i] = instructionNode;
      network.addNode(instructionNode);

      // Connect the instruction node output to the output PCollection node
      for (int j = 0; j < outputs.size(); ++j) {
        InstructionOutput instructionOutput = outputs.get(j);
        InstructionOutputNode outputNode =
            InstructionOutputNode.create(
                instructionOutput, "generatedPcollection" + this.idGenerator.getId());
        network.addNode(outputNode);
        if (parallelInstruction.getParDo() != null) {
          network.addEdge(
              instructionNode,
              outputNode,
              MultiOutputInfoEdge.create(
                  parallelInstruction.getParDo().getMultiOutputInfos().get(j)));
        } else {
          network.addEdge(instructionNode, outputNode, DefaultEdge.create());
        }
        outputNodes[i][j] = outputNode;
      }
    }

    // Connect PCollections as inputs to instructions
    for (ParallelInstructionNode instructionNode : instructionNodes) {
      ParallelInstruction parallelInstruction = instructionNode.getParallelInstruction();
      if (parallelInstruction.getFlatten() != null) {
        for (InstructionInput input :
            Apiary.listOrEmpty(parallelInstruction.getFlatten().getInputs())) {
          attachInput(input, network, instructionNode, outputNodes);
        }
      } else if (parallelInstruction.getParDo() != null) {
        attachInput(
            parallelInstruction.getParDo().getInput(), network, instructionNode, outputNodes);
      } else if (parallelInstruction.getPartialGroupByKey() != null) {
        attachInput(
            parallelInstruction.getPartialGroupByKey().getInput(),
            network,
            instructionNode,
            outputNodes);
      } else if (parallelInstruction.getRead() != null) {
        // Reads have no inputs so nothing to do
      } else if (parallelInstruction.getWrite() != null) {
        attachInput(
            parallelInstruction.getWrite().getInput(), network, instructionNode, outputNodes);
      } else {
        throw new IllegalArgumentException(
            String.format(
                "Unknown type of instruction %s for map task %s", parallelInstruction, mapTask));
      }
    }
    return network;
  }
@Test
  public void testParDo() {
    InstructionOutput readOutput = createInstructionOutput("Read.out");
    ParallelInstruction read = createParallelInstruction("Read", readOutput);
    read.setRead(new ReadInstruction());

    MultiOutputInfo parDoMultiOutput = createMultiOutputInfo("output");
    ParDoInstruction parDoInstruction = new ParDoInstruction();
    parDoInstruction.setInput(createInstructionInput(0, 0)); // Read.out
    parDoInstruction.setMultiOutputInfos(ImmutableList.of(parDoMultiOutput));
    InstructionOutput parDoOutput = createInstructionOutput("ParDo.out");
    ParallelInstruction parDo = createParallelInstruction("ParDo", parDoOutput);
    parDo.setParDo(parDoInstruction);

    MapTask mapTask = new MapTask();
    mapTask.setInstructions(ImmutableList.of(read, parDo));
    mapTask.setFactory(Transport.getJsonFactory());

    Network<Node, Edge> network =
        new MapTaskToNetworkFunction(IdGenerators.decrementingLongs()).apply(mapTask);
    assertNetworkProperties(network);
    assertEquals(4, network.nodes().size());
    assertEquals(3, network.edges().size());

    ParallelInstructionNode readNode = get(network, read);
    InstructionOutputNode readOutputNode = getOnlySuccessor(network, readNode);
    assertEquals(readOutput, readOutputNode.getInstructionOutput());

    ParallelInstructionNode parDoNode = getOnlySuccessor(network, readOutputNode);
    InstructionOutputNode parDoOutputNode = getOnlySuccessor(network, parDoNode);
    assertEquals(parDoOutput, parDoOutputNode.getInstructionOutput());

    assertEquals(
        parDoMultiOutput,
        ((MultiOutputInfoEdge)
                Iterables.getOnlyElement(network.edgesConnecting(parDoNode, parDoOutputNode)))
            .getMultiOutputInfo());
  }