public static MutableNetwork<Node, Edge> andReplaceForRunnerNetwork(
      MutableNetwork<Node, Edge> network) {
    Networks.replaceDirectedNetworkNodes(network, andReplaceForInstructionOutputNode());
    Networks.replaceDirectedNetworkNodes(network, andReplaceForParallelInstructionNode());
    return network;
  }
@Test
  public void testLengthPrefixAndReplaceForRunnerNetwork() throws Exception {
    Node readNode = createReadNode("Read", "Source", windowedValueCoder);
    Edge readNodeEdge = DefaultEdge.create();
    Node readNodeOut = createInstructionOutputNode("Read.out", windowedValueCoder);

    MutableNetwork<Node, Edge> network = createEmptyNetwork();
    network.addNode(readNode);
    network.addNode(readNodeOut);
    network.addEdge(readNode, readNodeOut, readNodeEdge);

    ParallelInstructionNode prefixedReadNode =
        createReadNode("Read", "Source", prefixedAndReplacedWindowedValueCoder);
    InstructionOutputNode prefixedReadNodeOut =
        createInstructionOutputNode("Read.out", prefixedAndReplacedWindowedValueCoder);

    MutableNetwork<Node, Edge> prefixedNetwork = andReplaceForRunnerNetwork(network);

    ImmutableSet.Builder<GenericJson> prefixedInstructions = ImmutableSet.builder();
    for (Node node : prefixedNetwork.nodes()) {
      if (node instanceof ParallelInstructionNode) {
        prefixedInstructions.add(((ParallelInstructionNode) node).getParallelInstruction());
      } else if (node instanceof InstructionOutputNode) {
        prefixedInstructions.add(((InstructionOutputNode) node).getInstructionOutput());
      }
    }

    assertThat(
        prefixedInstructions.build(),
        containsInAnyOrder(
            jsonOf(prefixedReadNodeOut.getInstructionOutput()),
            jsonOf(prefixedReadNode.getParallelInstruction())));
  }