LengthPrefixUnknownCoders { public static MutableNetwork<Node, Edge> andReplaceForRunnerNetwork( MutableNetwork<Node, Edge> network) { Networks.replaceDirectedNetworkNodes(network, andReplaceForInstructionOutputNode()); Networks.replaceDirectedNetworkNodes(network, andReplaceForParallelInstructionNode()); return network; }  static MutableNetwork<Node, Edge> andReplaceForRunnerNetwork(
      MutableNetwork<Node, Edge> network); static MutableNetwork<Node, Edge> forSdkNetwork(MutableNetwork<Node, Edge> network); static List<SideInputInfo> forSideInputInfos(
      List<SideInputInfo> sideInputInfos, boolean replaceWithByteArrayCoder);  }
@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()))); }