CreateRegisterFnOperationFunction implements Function<MutableNetwork<Node, Edge>, MutableNetwork<Node, Edge>> { @Override public MutableNetwork<Node, Edge> apply(MutableNetwork<Node, Edge> network) { Set<Node> runnerRootNodes = new HashSet<>(); Set<Node> sdkNodes = new HashSet<>(); Set<Node> sdkRootNodes = new HashSet<>(); for (ParallelInstructionNode node : Iterables.filter(network.nodes(), ParallelInstructionNode.class)) { if (executesInSdkHarness(node)) { sdkNodes.add(node); if (network.inDegree(node) == 0) { sdkRootNodes.add(node); } } else if (network.inDegree(node) == 0) { runnerRootNodes.add(node); } } if (sdkNodes.isEmpty()) { return network; } Set<Node> runnerToSdkBoundaries = new HashSet<>(); Set<Node> sdkToRunnerBoundaries = new HashSet<>(); ImmutableNetwork<Node, Edge> originalNetwork = ImmutableNetwork.copyOf(network); for (InstructionOutputNode outputNode : Iterables.filter(originalNetwork.nodes(), InstructionOutputNode.class)) { Set<Node> predecessorRunnerInstructions = new HashSet<>(); Set<Node> predecessorSdkInstructions = new HashSet<>(); for (Node predecessorInstruction : originalNetwork.predecessors(outputNode)) { if (sdkNodes.contains(predecessorInstruction)) { predecessorSdkInstructions.add(predecessorInstruction); } else { predecessorRunnerInstructions.add(predecessorInstruction); } } Set<Node> successorRunnerInstructions = new HashSet<>(); Set<Node> successorSdkInstructions = new HashSet<>(); for (Node successorInstruction : originalNetwork.successors(outputNode)) { if (sdkNodes.contains(successorInstruction)) { successorSdkInstructions.add(successorInstruction); } else { successorRunnerInstructions.add(successorInstruction); } } if (!predecessorRunnerInstructions.isEmpty() && !successorSdkInstructions.isEmpty()) { runnerToSdkBoundaries.add( rewireAcrossSdkRunnerPortNode( network, outputNode, predecessorRunnerInstructions, successorSdkInstructions)); } if (!predecessorSdkInstructions.isEmpty() && !successorRunnerInstructions.isEmpty()) { sdkToRunnerBoundaries.add( rewireAcrossSdkRunnerPortNode( network, outputNode, predecessorSdkInstructions, successorRunnerInstructions)); } if (network.inDegree(outputNode) == 0) { network.removeNode(outputNode); } } Set<Node> allRunnerNodes = Networks.reachableNodes( network, Sets.union(runnerRootNodes, sdkToRunnerBoundaries), runnerToSdkBoundaries); if (this.useExecutableStageBundleExecution) { allRunnerNodes = Sets.difference(allRunnerNodes, Sets.union(runnerToSdkBoundaries, sdkToRunnerBoundaries)); } MutableNetwork<Node, Edge> runnerNetwork = Graphs.inducedSubgraph(network, allRunnerNodes); for (Node sdkRoot : Sets.union(sdkRootNodes, runnerToSdkBoundaries)) { Set<Node> sdkSubnetworkNodes = Networks.reachableNodes(network, ImmutableSet.of(sdkRoot), sdkToRunnerBoundaries); MutableNetwork<Node, Edge> sdkNetwork = Graphs.inducedSubgraph(network, sdkSubnetworkNodes); Node registerFnNode = registerFnOperationFunction.apply(sdkNetwork); runnerNetwork.addNode(registerFnNode); if (this.useExecutableStageBundleExecution) { for (Node predecessor : Sets.intersection(sdkSubnetworkNodes, runnerToSdkBoundaries)) { predecessor = network.predecessors(predecessor).iterator().next(); runnerNetwork.addEdge(predecessor, registerFnNode, HappensBeforeEdge.create()); } for (Node successor : Sets.intersection(sdkSubnetworkNodes, sdkToRunnerBoundaries)) { successor = network.successors(successor).iterator().next(); runnerNetwork.addEdge(registerFnNode, successor, HappensBeforeEdge.create()); } } else { for (Node predecessor : Sets.intersection(sdkSubnetworkNodes, runnerToSdkBoundaries)) { runnerNetwork.addEdge(predecessor, registerFnNode, HappensBeforeEdge.create()); } for (Node successor : Sets.intersection(sdkSubnetworkNodes, sdkToRunnerBoundaries)) { runnerNetwork.addEdge(registerFnNode, successor, HappensBeforeEdge.create()); } } } return runnerNetwork; } CreateRegisterFnOperationFunction(
      IdGenerator idGenerator,
      Supplier<Node> portSupplier,
      Function<MutableNetwork<Node, Edge>, Node> registerFnOperationFunction,
      boolean useExecutableStageBundleExecution); @Override MutableNetwork<Node, Edge> apply(MutableNetwork<Node, Edge> network);  }
@Test public void testEmptyGraph() { MutableNetwork<Node, Edge> appliedNetwork = createRegisterFnOperation.apply(createEmptyNetwork()); MutableNetwork<Node, Edge> expectedNetwork = createEmptyNetwork(); assertNetworkMaintainsBipartiteStructure(appliedNetwork); assertEquals( String.format("Expected network %s but got network %s", expectedNetwork, appliedNetwork), expectedNetwork, appliedNetwork); }
@Test public void testAllRunnerGraph() { Node readNode = createReadNode("Read", Nodes.ExecutionLocation.RUNNER_HARNESS); Edge readNodeEdge = DefaultEdge.create(); Node readNodeOut = createInstructionOutputNode("Read.out"); Edge readNodeOutEdge = DefaultEdge.create(); Node parDoNode = createParDoNode("ParDo", Nodes.ExecutionLocation.RUNNER_HARNESS); Edge parDoNodeEdge = DefaultEdge.create(); Node parDoNodeOut = createInstructionOutputNode("ParDo.out"); MutableNetwork<Node, Edge> expectedNetwork = createEmptyNetwork(); expectedNetwork.addNode(readNode); expectedNetwork.addNode(readNodeOut); expectedNetwork.addNode(parDoNode); expectedNetwork.addNode(parDoNodeOut); expectedNetwork.addEdge(readNode, readNodeOut, readNodeEdge); expectedNetwork.addEdge(readNodeOut, parDoNode, readNodeOutEdge); expectedNetwork.addEdge(parDoNode, parDoNodeOut, parDoNodeEdge); MutableNetwork<Node, Edge> appliedNetwork = createRegisterFnOperation.apply(Graphs.copyOf(expectedNetwork)); assertNetworkMaintainsBipartiteStructure(appliedNetwork); assertEquals( String.format("Expected network %s but got network %s", expectedNetwork, appliedNetwork), expectedNetwork, appliedNetwork); }
@Test public void testAllSdkGraph() { Node sdkPortionNode = TestNode.create("SdkPortion"); @SuppressWarnings({"unchecked", "rawtypes"}) ArgumentCaptor<MutableNetwork<Node, Edge>> networkCapture = ArgumentCaptor.forClass((Class) MutableNetwork.class); when(registerFnOperationFunction.apply(networkCapture.capture())).thenReturn(sdkPortionNode); Node readNode = createReadNode("Read", Nodes.ExecutionLocation.SDK_HARNESS); Edge readNodeEdge = DefaultEdge.create(); Node readNodeOut = createInstructionOutputNode("Read.out"); Edge readNodeOutEdge = DefaultEdge.create(); Node parDoNode = createParDoNode("ParDo", Nodes.ExecutionLocation.SDK_HARNESS); Edge parDoNodeEdge = DefaultEdge.create(); Node parDoNodeOut = createInstructionOutputNode("ParDo.out"); MutableNetwork<Node, Edge> network = createEmptyNetwork(); network.addNode(readNode); network.addNode(readNodeOut); network.addNode(parDoNode); network.addNode(parDoNodeOut); network.addEdge(readNode, readNodeOut, readNodeEdge); network.addEdge(readNodeOut, parDoNode, readNodeOutEdge); network.addEdge(parDoNode, parDoNodeOut, parDoNodeEdge); MutableNetwork<Node, Edge> expectedNetwork = createEmptyNetwork(); expectedNetwork.addNode(sdkPortionNode); MutableNetwork<Node, Edge> appliedNetwork = createRegisterFnOperation.apply(Graphs.copyOf(network)); assertNetworkMaintainsBipartiteStructure(appliedNetwork); assertNetworkMaintainsBipartiteStructure(networkCapture.getValue()); assertEquals( String.format("Expected network %s but got network %s", expectedNetwork, appliedNetwork), expectedNetwork, appliedNetwork); assertEquals( String.format("Expected network %s but got network %s", network, networkCapture.getValue()), network, networkCapture.getValue()); }
@Test public void testRunnerToSdkToRunnerGraph() { Node sdkPortion = TestNode.create("SdkPortion"); @SuppressWarnings({"unchecked", "rawtypes"}) ArgumentCaptor<MutableNetwork<Node, Edge>> networkCapture = ArgumentCaptor.forClass((Class) MutableNetwork.class); when(registerFnOperationFunction.apply(networkCapture.capture())).thenReturn(sdkPortion); Node firstPort = TestNode.create("FirstPort"); Node secondPort = TestNode.create("SecondPort"); when(portSupplier.get()).thenReturn(firstPort, secondPort); Node readNode = createReadNode("Read", Nodes.ExecutionLocation.RUNNER_HARNESS); Edge readNodeEdge = DefaultEdge.create(); Node readNodeOut = createInstructionOutputNode("Read.out"); Edge readNodeOutEdge = DefaultEdge.create(); Node sdkParDoNode = createParDoNode("SdkParDo", Nodes.ExecutionLocation.SDK_HARNESS); Edge sdkParDoNodeEdge = DefaultEdge.create(); Node sdkParDoNodeOut = createInstructionOutputNode("SdkParDo.out"); Edge sdkParDoNodeOutEdge = DefaultEdge.create(); Node runnerParDoNode = createParDoNode("RunnerParDo", Nodes.ExecutionLocation.RUNNER_HARNESS); Edge runnerParDoNodeEdge = DefaultEdge.create(); Node runnerParDoNodeOut = createInstructionOutputNode("RunnerParDo.out"); MutableNetwork<Node, Edge> network = createEmptyNetwork(); network.addNode(readNode); network.addNode(readNodeOut); network.addNode(sdkParDoNodeOut); network.addNode(sdkParDoNodeOut); network.addNode(runnerParDoNode); network.addNode(runnerParDoNodeOut); network.addEdge(readNode, readNodeOut, readNodeEdge); network.addEdge(readNodeOut, sdkParDoNode, readNodeOutEdge); network.addEdge(sdkParDoNode, sdkParDoNodeOut, sdkParDoNodeEdge); network.addEdge(sdkParDoNodeOut, runnerParDoNode, sdkParDoNodeOutEdge); network.addEdge(runnerParDoNode, runnerParDoNodeOut, runnerParDoNodeEdge); MutableNetwork<Node, Edge> appliedNetwork = createRegisterFnOperation.apply(Graphs.copyOf(network)); assertNetworkMaintainsBipartiteStructure(appliedNetwork); Node newOutA = Iterables.getOnlyElement(appliedNetwork.predecessors(firstPort)); Node newOutB = Iterables.getOnlyElement(appliedNetwork.successors(secondPort)); assertThat( appliedNetwork.nodes(), containsInAnyOrder( readNode, newOutA, firstPort, sdkPortion, secondPort, newOutB, runnerParDoNode, runnerParDoNodeOut)); assertThat(appliedNetwork.successors(readNode), containsInAnyOrder(newOutA)); assertThat(appliedNetwork.successors(newOutA), containsInAnyOrder(firstPort)); assertThat(appliedNetwork.successors(firstPort), containsInAnyOrder(sdkPortion)); assertThat(appliedNetwork.successors(sdkPortion), containsInAnyOrder(secondPort)); assertThat(appliedNetwork.successors(secondPort), containsInAnyOrder(newOutB)); assertThat(appliedNetwork.successors(newOutB), containsInAnyOrder(runnerParDoNode)); assertThat(appliedNetwork.successors(runnerParDoNode), containsInAnyOrder(runnerParDoNodeOut)); assertThat( appliedNetwork.edgesConnecting(firstPort, sdkPortion), everyItem(Matchers.<Edges.Edge>instanceOf(HappensBeforeEdge.class))); assertThat( appliedNetwork.edgesConnecting(sdkPortion, secondPort), everyItem(Matchers.<Edges.Edge>instanceOf(HappensBeforeEdge.class))); MutableNetwork<Node, Edge> sdkSubnetwork = networkCapture.getValue(); assertNetworkMaintainsBipartiteStructure(sdkSubnetwork); Node sdkNewOutA = Iterables.getOnlyElement(sdkSubnetwork.successors(firstPort)); Node sdkNewOutB = Iterables.getOnlyElement(sdkSubnetwork.predecessors(secondPort)); assertThat( sdkSubnetwork.nodes(), containsInAnyOrder(firstPort, sdkNewOutA, sdkParDoNode, sdkNewOutB, secondPort)); assertThat(sdkSubnetwork.successors(firstPort), containsInAnyOrder(sdkNewOutA)); assertThat(sdkSubnetwork.successors(sdkNewOutA), containsInAnyOrder(sdkParDoNode)); assertThat(sdkSubnetwork.successors(sdkParDoNode), containsInAnyOrder(sdkNewOutB)); assertThat(sdkSubnetwork.successors(sdkNewOutB), containsInAnyOrder(secondPort)); }
@Test public void testSdkToRunnerToSdkGraph() { Node firstSdkPortion = TestNode.create("FirstSdkPortion"); Node secondSdkPortion = TestNode.create("SecondSdkPortion"); @SuppressWarnings({"unchecked", "rawtypes"}) ArgumentCaptor<MutableNetwork<Node, Edge>> networkCapture = ArgumentCaptor.forClass((Class) MutableNetwork.class); when(registerFnOperationFunction.apply(networkCapture.capture())) .thenReturn(firstSdkPortion, secondSdkPortion); Node firstPort = TestNode.create("FirstPort"); Node secondPort = TestNode.create("SecondPort"); when(portSupplier.get()).thenReturn(firstPort, secondPort); Node readNode = createReadNode("Read", Nodes.ExecutionLocation.SDK_HARNESS); Edge readNodeEdge = DefaultEdge.create(); Node readNodeOut = createInstructionOutputNode("Read.out"); Edge readNodeOutEdge = DefaultEdge.create(); Node runnerParDoNode = createParDoNode("RunnerParDo", Nodes.ExecutionLocation.RUNNER_HARNESS); Edge runnerParDoNodeEdge = DefaultEdge.create(); Node runnerParDoNodeOut = createInstructionOutputNode("RunnerParDo.out"); Edge runnerParDoNodeOutEdge = DefaultEdge.create(); Node sdkParDoNode = createParDoNode("SdkParDo", Nodes.ExecutionLocation.SDK_HARNESS); Edge sdkParDoNodeEdge = DefaultEdge.create(); Node sdkParDoNodeOut = createInstructionOutputNode("SdkParDo.out"); MutableNetwork<Node, Edge> network = createEmptyNetwork(); network.addNode(readNode); network.addNode(readNodeOut); network.addNode(runnerParDoNode); network.addNode(runnerParDoNodeOut); network.addNode(sdkParDoNodeOut); network.addNode(sdkParDoNodeOut); network.addEdge(readNode, readNodeOut, readNodeEdge); network.addEdge(readNodeOut, runnerParDoNode, readNodeOutEdge); network.addEdge(runnerParDoNode, runnerParDoNodeOut, runnerParDoNodeEdge); network.addEdge(runnerParDoNodeOut, sdkParDoNode, runnerParDoNodeOutEdge); network.addEdge(sdkParDoNode, sdkParDoNodeOut, sdkParDoNodeEdge); MutableNetwork<Node, Edge> appliedNetwork = createRegisterFnOperation.apply(Graphs.copyOf(network)); assertNetworkMaintainsBipartiteStructure(appliedNetwork); Node newOutA = Iterables.getOnlyElement(appliedNetwork.successors(firstPort)); Node newOutB = Iterables.getOnlyElement(appliedNetwork.predecessors(secondPort)); assertThat( appliedNetwork.nodes(), containsInAnyOrder( firstSdkPortion, firstPort, newOutA, runnerParDoNode, newOutB, secondPort, secondSdkPortion)); assertThat(appliedNetwork.successors(firstSdkPortion), containsInAnyOrder(firstPort)); assertThat(appliedNetwork.successors(firstPort), containsInAnyOrder(newOutA)); assertThat(appliedNetwork.successors(newOutA), containsInAnyOrder(runnerParDoNode)); assertThat(appliedNetwork.successors(runnerParDoNode), containsInAnyOrder(newOutB)); assertThat(appliedNetwork.successors(newOutB), containsInAnyOrder(secondPort)); assertThat(appliedNetwork.successors(secondPort), containsInAnyOrder(secondSdkPortion)); assertThat( appliedNetwork.edgesConnecting(firstSdkPortion, firstPort), everyItem(Matchers.<Edges.Edge>instanceOf(HappensBeforeEdge.class))); assertThat( appliedNetwork.edgesConnecting(secondPort, secondSdkPortion), everyItem(Matchers.<Edges.Edge>instanceOf(HappensBeforeEdge.class))); List<MutableNetwork<Node, Edge>> sdkSubnetworks = networkCapture.getAllValues(); MutableNetwork<Node, Edge> firstSdkSubnetwork; MutableNetwork<Node, Edge> secondSdkSubnetwork; if (sdkSubnetworks.get(0).nodes().contains(readNode)) { firstSdkSubnetwork = sdkSubnetworks.get(0); secondSdkSubnetwork = sdkSubnetworks.get(1); } else { firstSdkSubnetwork = sdkSubnetworks.get(1); secondSdkSubnetwork = sdkSubnetworks.get(0); } assertNetworkMaintainsBipartiteStructure(firstSdkSubnetwork); assertNetworkMaintainsBipartiteStructure(secondSdkSubnetwork); Node sdkNewOutA = Iterables.getOnlyElement(firstSdkSubnetwork.predecessors(firstPort)); assertThat(firstSdkSubnetwork.nodes(), containsInAnyOrder(readNode, sdkNewOutA, firstPort)); assertThat(firstSdkSubnetwork.successors(readNode), containsInAnyOrder(sdkNewOutA)); assertThat(firstSdkSubnetwork.successors(sdkNewOutA), containsInAnyOrder(firstPort)); Node sdkNewOutB = Iterables.getOnlyElement(secondSdkSubnetwork.successors(secondPort)); assertThat( secondSdkSubnetwork.nodes(), containsInAnyOrder(secondPort, sdkNewOutB, sdkParDoNode, sdkParDoNodeOut)); assertThat(secondSdkSubnetwork.successors(secondPort), containsInAnyOrder(sdkNewOutB)); assertThat(secondSdkSubnetwork.successors(sdkNewOutB), containsInAnyOrder(sdkParDoNode)); assertThat(secondSdkSubnetwork.successors(sdkParDoNode), containsInAnyOrder(sdkParDoNodeOut)); }
@Test public void testRunnerAndSdkToRunnerAndSdkGraph() { Node firstSdkPortion = TestNode.create("FirstSdkPortion"); Node secondSdkPortion = TestNode.create("SecondSdkPortion"); @SuppressWarnings({"unchecked", "rawtypes"}) ArgumentCaptor<MutableNetwork<Node, Edge>> networkCapture = ArgumentCaptor.forClass((Class) MutableNetwork.class); when(registerFnOperationFunction.apply(networkCapture.capture())) .thenReturn(firstSdkPortion, secondSdkPortion); Node firstPort = TestNode.create("FirstPort"); Node secondPort = TestNode.create("SecondPort"); when(portSupplier.get()).thenReturn(firstPort, secondPort); Node runnerReadNode = createReadNode("RunnerRead", Nodes.ExecutionLocation.RUNNER_HARNESS); Edge runnerReadNodeEdge = DefaultEdge.create(); Node sdkReadNode = createReadNode("SdkRead", Nodes.ExecutionLocation.SDK_HARNESS); Edge sdkReadNodeEdge = DefaultEdge.create(); Node readNodeOut = createInstructionOutputNode("Read.out"); Edge readNodeOutToRunnerEdge = DefaultEdge.create(); Edge readNodeOutToSdkEdge = DefaultEdge.create(); Node runnerParDoNode = createParDoNode("RunnerParDo", Nodes.ExecutionLocation.RUNNER_HARNESS); Edge runnerParDoNodeEdge = DefaultEdge.create(); Node runnerParDoNodeOut = createInstructionOutputNode("RunnerParDo.out"); Node sdkParDoNode = createParDoNode("SdkParDo", Nodes.ExecutionLocation.SDK_HARNESS); Edge sdkParDoNodeEdge = DefaultEdge.create(); Node sdkParDoNodeOut = createInstructionOutputNode("SdkParDo.out"); MutableNetwork<Node, Edge> network = createEmptyNetwork(); network.addNode(sdkReadNode); network.addNode(runnerReadNode); network.addNode(readNodeOut); network.addNode(runnerParDoNode); network.addNode(runnerParDoNodeOut); network.addNode(sdkParDoNodeOut); network.addNode(sdkParDoNodeOut); network.addEdge(sdkReadNode, readNodeOut, sdkReadNodeEdge); network.addEdge(runnerReadNode, readNodeOut, runnerReadNodeEdge); network.addEdge(readNodeOut, runnerParDoNode, readNodeOutToRunnerEdge); network.addEdge(readNodeOut, sdkParDoNode, readNodeOutToSdkEdge); network.addEdge(runnerParDoNode, runnerParDoNodeOut, runnerParDoNodeEdge); network.addEdge(sdkParDoNode, sdkParDoNodeOut, sdkParDoNodeEdge); MutableNetwork<Node, Edge> appliedNetwork = createRegisterFnOperation.apply(Graphs.copyOf(network)); assertNetworkMaintainsBipartiteStructure(appliedNetwork); Node sdkPortionA; Node sdkPortionB; if (appliedNetwork.inDegree(firstSdkPortion) == 0) { sdkPortionA = firstSdkPortion; sdkPortionB = secondSdkPortion; } else { sdkPortionA = secondSdkPortion; sdkPortionB = firstSdkPortion; } Node portA = Iterables.getOnlyElement(appliedNetwork.successors(sdkPortionA)); Node portB = Iterables.getOnlyElement(appliedNetwork.predecessors(sdkPortionB)); Node newOutA = Iterables.getOnlyElement(appliedNetwork.successors(portA)); Node newOutB = Iterables.getOnlyElement(appliedNetwork.predecessors(portB)); assertThat( appliedNetwork.nodes(), containsInAnyOrder( runnerReadNode, firstSdkPortion, secondSdkPortion, portA, newOutA, portB, newOutB, runnerParDoNode, runnerParDoNodeOut)); assertThat(appliedNetwork.successors(runnerReadNode), containsInAnyOrder(newOutB)); assertThat(appliedNetwork.successors(newOutB), containsInAnyOrder(runnerParDoNode, portB)); assertThat(appliedNetwork.successors(portB), containsInAnyOrder(sdkPortionB)); assertThat(appliedNetwork.successors(sdkPortionA), containsInAnyOrder(portA)); assertThat(appliedNetwork.successors(portA), containsInAnyOrder(newOutA)); assertThat(appliedNetwork.successors(newOutA), containsInAnyOrder(runnerParDoNode)); assertThat(appliedNetwork.successors(runnerParDoNode), containsInAnyOrder(runnerParDoNodeOut)); assertThat( appliedNetwork.edgesConnecting(sdkPortionA, portA), everyItem(Matchers.<Edges.Edge>instanceOf(HappensBeforeEdge.class))); assertThat( appliedNetwork.edgesConnecting(portB, sdkPortionB), everyItem(Matchers.<Edges.Edge>instanceOf(HappensBeforeEdge.class))); List<MutableNetwork<Node, Edge>> sdkSubnetworks = networkCapture.getAllValues(); MutableNetwork<Node, Edge> sdkSubnetworkA; MutableNetwork<Node, Edge> sdkSubnetworkB; if (sdkSubnetworks.get(0).nodes().contains(sdkReadNode)) { sdkSubnetworkA = sdkSubnetworks.get(0); sdkSubnetworkB = sdkSubnetworks.get(1); } else { sdkSubnetworkA = sdkSubnetworks.get(1); sdkSubnetworkB = sdkSubnetworks.get(0); } assertNetworkMaintainsBipartiteStructure(sdkSubnetworkA); assertNetworkMaintainsBipartiteStructure(sdkSubnetworkB); Node sdkNewOutA = Iterables.getOnlyElement(sdkSubnetworkA.predecessors(portA)); assertThat( sdkSubnetworkA.nodes(), containsInAnyOrder(sdkReadNode, portA, sdkNewOutA, sdkParDoNode, sdkParDoNodeOut)); assertThat(sdkSubnetworkA.successors(sdkReadNode), containsInAnyOrder(sdkNewOutA)); assertThat(sdkSubnetworkA.successors(sdkNewOutA), containsInAnyOrder(portA, sdkParDoNode)); assertThat(sdkSubnetworkA.successors(sdkParDoNode), containsInAnyOrder(sdkParDoNodeOut)); Node sdkNewOutB = Iterables.getOnlyElement(sdkSubnetworkB.successors(portB)); assertThat( sdkSubnetworkB.nodes(), containsInAnyOrder(portB, sdkNewOutB, sdkParDoNode, sdkParDoNodeOut)); assertThat(sdkSubnetworkB.successors(portB), containsInAnyOrder(sdkNewOutB)); assertThat(sdkSubnetworkB.successors(sdkNewOutB), containsInAnyOrder(sdkParDoNode)); assertThat(sdkSubnetworkB.successors(sdkParDoNode), containsInAnyOrder(sdkParDoNodeOut)); }