@Test public void testConcatenate() throws URISyntaxException { Dag<JobExecutionPlan> dag1 = buildDag(2, 0, false); Dag<JobExecutionPlan> dag2 = buildDag(2, 2, false); Dag<JobExecutionPlan> dagNew = FlowGraphPath.concatenate(dag1, dag2); Assert.assertEquals(dagNew.getStartNodes().size(), 1); Assert.assertEquals(dagNew.getEndNodes().size(), 1); Assert.assertEquals(dagNew.getNodes().size(), 4); Assert.assertEquals(dagNew.getStartNodes().get(0).getValue().getJobSpec().getConfig().getString(ConfigurationKeys.JOB_NAME_KEY), "job0"); Assert.assertEquals(dagNew.getEndNodes().get(0).getValue().getJobSpec().getConfig().getString(ConfigurationKeys.JOB_NAME_KEY), "job3"); Assert.assertEquals(dagNew.getEndNodes().get(0).getValue().getJobSpec().getConfig().getString(ConfigurationKeys.JOB_DEPENDENCIES), "job2"); dag1 = buildDag(1, 0, true); dag2 = buildDag(2, 1, false); dagNew = FlowGraphPath.concatenate(dag1, dag2); Assert.assertEquals(dagNew.getStartNodes().size(), 2); Assert.assertEquals(dagNew.getEndNodes().size(), 2); Assert.assertEquals(dagNew.getNodes().size(), 3); Assert.assertEquals(dagNew.getStartNodes().get(0).getValue().getJobSpec().getConfig().getString(ConfigurationKeys.JOB_NAME_KEY), "job0"); Assert.assertEquals(dagNew.getStartNodes().get(1).getValue().getJobSpec().getConfig().getString(ConfigurationKeys.JOB_NAME_KEY), "job1"); Assert.assertEquals(dagNew.getEndNodes().get(0).getValue().getJobSpec().getConfig().getString(ConfigurationKeys.JOB_NAME_KEY), "job2"); Assert.assertEquals(dagNew.getEndNodes().get(0).getValue().getJobSpec().getConfig().getString(ConfigurationKeys.JOB_DEPENDENCIES), "job1"); Assert.assertEquals(dagNew.getEndNodes().get(1).getValue().getJobSpec().getConfig().getString(ConfigurationKeys.JOB_NAME_KEY), "job0"); Assert.assertFalse(dagNew.getEndNodes().get(1).getValue().getJobSpec().getConfig().hasPath(ConfigurationKeys.JOB_DEPENDENCIES)); dag1 = buildDag(2, 0, true); dag2 = buildDag(2, 2, false); dagNew = FlowGraphPath.concatenate(dag1, dag2); Assert.assertEquals(dagNew.getStartNodes().size(), 1); Assert.assertEquals(dagNew.getEndNodes().size(), 2); Assert.assertEquals(dagNew.getNodes().size(), 4); Assert.assertEquals(dagNew.getStartNodes().get(0).getValue().getJobSpec().getConfig().getString(ConfigurationKeys.JOB_NAME_KEY), "job0"); Assert.assertEquals(dagNew.getEndNodes().get(0).getValue().getJobSpec().getConfig().getString(ConfigurationKeys.JOB_NAME_KEY), "job3"); Assert.assertEquals(dagNew.getEndNodes().get(0).getValue().getJobSpec().getConfig().getString(ConfigurationKeys.JOB_DEPENDENCIES), "job2"); Assert.assertEquals(dagNew.getEndNodes().get(1).getValue().getJobSpec().getConfig().getString(ConfigurationKeys.JOB_NAME_KEY), "job1"); Assert.assertEquals(dagNew.getEndNodes().get(1).getValue().getJobSpec().getConfig().getString(ConfigurationKeys.JOB_DEPENDENCIES), "job0"); dag1 = buildDag(1, 0, true); dag2 = buildDag(1, 1, false); dagNew = FlowGraphPath.concatenate(dag1, dag2); Assert.assertEquals(dagNew.getStartNodes().size(), 2); Assert.assertEquals(dagNew.getEndNodes().size(), 2); Assert.assertEquals(dagNew.getNodes().size(), 2); Assert.assertEquals(dagNew.getStartNodes().get(0).getValue().getJobSpec().getConfig().getString(ConfigurationKeys.JOB_NAME_KEY), "job0"); Assert.assertEquals(dagNew.getStartNodes().get(1).getValue().getJobSpec().getConfig().getString(ConfigurationKeys.JOB_NAME_KEY), "job1"); Assert.assertFalse(dagNew.getStartNodes().get(1).getValue().getJobSpec().getConfig().hasPath(ConfigurationKeys.JOB_DEPENDENCIES)); Assert.assertFalse(dagNew.getStartNodes().get(1).getValue().getJobSpec().getConfig().hasPath(ConfigurationKeys.JOB_DEPENDENCIES)); }
@VisibleForTesting static Dag<JobExecutionPlan> concatenate(Dag<JobExecutionPlan> dagLeft, Dag<JobExecutionPlan> dagRight) { Set<DagNode<JobExecutionPlan>> forkNodes = dagLeft.getEndNodes().stream(). filter(endNode -> isNodeForkable(endNode)).collect(Collectors.toSet()); Set<DagNode<JobExecutionPlan>> dependencyNodes = dagLeft.getDependencyNodes(forkNodes); if (!dependencyNodes.isEmpty()) { List<String> dependenciesList = dependencyNodes.stream() .map(dagNode -> dagNode.getValue().getJobSpec().getConfig().getString(ConfigurationKeys.JOB_NAME_KEY)) .collect(Collectors.toList()); String dependencies = Joiner.on(",").join(dependenciesList); for (DagNode<JobExecutionPlan> childNode : dagRight.getStartNodes()) { JobSpec jobSpec = childNode.getValue().getJobSpec(); jobSpec.setConfig(jobSpec.getConfig().withValue(ConfigurationKeys.JOB_DEPENDENCIES, ConfigValueFactory.fromAnyRef(dependencies))); } } return dagLeft.concatenate(dagRight, forkNodes); }
FlowGraphPath { @VisibleForTesting static Dag<JobExecutionPlan> concatenate(Dag<JobExecutionPlan> dagLeft, Dag<JobExecutionPlan> dagRight) { Set<DagNode<JobExecutionPlan>> forkNodes = dagLeft.getEndNodes().stream(). filter(endNode -> isNodeForkable(endNode)).collect(Collectors.toSet()); Set<DagNode<JobExecutionPlan>> dependencyNodes = dagLeft.getDependencyNodes(forkNodes); if (!dependencyNodes.isEmpty()) { List<String> dependenciesList = dependencyNodes.stream() .map(dagNode -> dagNode.getValue().getJobSpec().getConfig().getString(ConfigurationKeys.JOB_NAME_KEY)) .collect(Collectors.toList()); String dependencies = Joiner.on(",").join(dependenciesList); for (DagNode<JobExecutionPlan> childNode : dagRight.getStartNodes()) { JobSpec jobSpec = childNode.getValue().getJobSpec(); jobSpec.setConfig(jobSpec.getConfig().withValue(ConfigurationKeys.JOB_DEPENDENCIES, ConfigValueFactory.fromAnyRef(dependencies))); } } return dagLeft.concatenate(dagRight, forkNodes); } }
FlowGraphPath { @VisibleForTesting static Dag<JobExecutionPlan> concatenate(Dag<JobExecutionPlan> dagLeft, Dag<JobExecutionPlan> dagRight) { Set<DagNode<JobExecutionPlan>> forkNodes = dagLeft.getEndNodes().stream(). filter(endNode -> isNodeForkable(endNode)).collect(Collectors.toSet()); Set<DagNode<JobExecutionPlan>> dependencyNodes = dagLeft.getDependencyNodes(forkNodes); if (!dependencyNodes.isEmpty()) { List<String> dependenciesList = dependencyNodes.stream() .map(dagNode -> dagNode.getValue().getJobSpec().getConfig().getString(ConfigurationKeys.JOB_NAME_KEY)) .collect(Collectors.toList()); String dependencies = Joiner.on(",").join(dependenciesList); for (DagNode<JobExecutionPlan> childNode : dagRight.getStartNodes()) { JobSpec jobSpec = childNode.getValue().getJobSpec(); jobSpec.setConfig(jobSpec.getConfig().withValue(ConfigurationKeys.JOB_DEPENDENCIES, ConfigValueFactory.fromAnyRef(dependencies))); } } return dagLeft.concatenate(dagRight, forkNodes); } FlowGraphPath(FlowSpec flowSpec, Long flowExecutionId); }
FlowGraphPath { @VisibleForTesting static Dag<JobExecutionPlan> concatenate(Dag<JobExecutionPlan> dagLeft, Dag<JobExecutionPlan> dagRight) { Set<DagNode<JobExecutionPlan>> forkNodes = dagLeft.getEndNodes().stream(). filter(endNode -> isNodeForkable(endNode)).collect(Collectors.toSet()); Set<DagNode<JobExecutionPlan>> dependencyNodes = dagLeft.getDependencyNodes(forkNodes); if (!dependencyNodes.isEmpty()) { List<String> dependenciesList = dependencyNodes.stream() .map(dagNode -> dagNode.getValue().getJobSpec().getConfig().getString(ConfigurationKeys.JOB_NAME_KEY)) .collect(Collectors.toList()); String dependencies = Joiner.on(",").join(dependenciesList); for (DagNode<JobExecutionPlan> childNode : dagRight.getStartNodes()) { JobSpec jobSpec = childNode.getValue().getJobSpec(); jobSpec.setConfig(jobSpec.getConfig().withValue(ConfigurationKeys.JOB_DEPENDENCIES, ConfigValueFactory.fromAnyRef(dependencies))); } } return dagLeft.concatenate(dagRight, forkNodes); } FlowGraphPath(FlowSpec flowSpec, Long flowExecutionId); void addPath(List<FlowEdgeContext> path); Dag<JobExecutionPlan> asDag(Config sysConfig); }
FlowGraphPath { @VisibleForTesting static Dag<JobExecutionPlan> concatenate(Dag<JobExecutionPlan> dagLeft, Dag<JobExecutionPlan> dagRight) { Set<DagNode<JobExecutionPlan>> forkNodes = dagLeft.getEndNodes().stream(). filter(endNode -> isNodeForkable(endNode)).collect(Collectors.toSet()); Set<DagNode<JobExecutionPlan>> dependencyNodes = dagLeft.getDependencyNodes(forkNodes); if (!dependencyNodes.isEmpty()) { List<String> dependenciesList = dependencyNodes.stream() .map(dagNode -> dagNode.getValue().getJobSpec().getConfig().getString(ConfigurationKeys.JOB_NAME_KEY)) .collect(Collectors.toList()); String dependencies = Joiner.on(",").join(dependenciesList); for (DagNode<JobExecutionPlan> childNode : dagRight.getStartNodes()) { JobSpec jobSpec = childNode.getValue().getJobSpec(); jobSpec.setConfig(jobSpec.getConfig().withValue(ConfigurationKeys.JOB_DEPENDENCIES, ConfigValueFactory.fromAnyRef(dependencies))); } } return dagLeft.concatenate(dagRight, forkNodes); } FlowGraphPath(FlowSpec flowSpec, Long flowExecutionId); void addPath(List<FlowEdgeContext> path); Dag<JobExecutionPlan> asDag(Config sysConfig);  }