@Test public void testCreateDagWithDuplicateJobNames() throws Exception { Config flowConfig1 = ConfigBuilder.create().addPrimitive(ConfigurationKeys.FLOW_NAME_KEY, "flowName") .addPrimitive(ConfigurationKeys.FLOW_GROUP_KEY, "flowGroup").build(); Config flowConfig2 = ConfigBuilder.create().addPrimitive(ConfigurationKeys.FLOW_NAME_KEY, "flowName") .addPrimitive(ConfigurationKeys.FLOW_GROUP_KEY, "flowGroup").build(); Config flowConfig3 = ConfigBuilder.create().addPrimitive(ConfigurationKeys.FLOW_NAME_KEY, "flowName") .addPrimitive(ConfigurationKeys.FLOW_GROUP_KEY, "flowGroup").build(); List<Config> flowConfigs = Arrays.asList(flowConfig1, flowConfig2, flowConfig3); Config jobConfig1 = ConfigBuilder.create().addPrimitive(ConfigurationKeys.JOB_NAME_KEY, "job1") .addPrimitive(FlowGraphConfigurationKeys.FLOW_EDGE_ID_KEY, "source:destination:edgeName1").build(); Config jobConfig2 = ConfigBuilder.create().addPrimitive(ConfigurationKeys.JOB_NAME_KEY, "job2") .addPrimitive(FlowGraphConfigurationKeys.FLOW_EDGE_ID_KEY, "source:destination:edgeName2").build(); Config jobConfig3 = ConfigBuilder.create().addPrimitive(ConfigurationKeys.JOB_NAME_KEY, "job1") .addPrimitive(FlowGraphConfigurationKeys.FLOW_EDGE_ID_KEY, "source:destination:edgeName3").build(); List<Config> jobConfigs = Arrays.asList(jobConfig1, jobConfig2, jobConfig3); List<JobExecutionPlan> jobExecutionPlans = new ArrayList<>(); for (int i = 0; i < 3; i++) { Config jobConfig = jobConfigs.get(i); if (i > 0) { String previousJobName = jobExecutionPlans.get(jobExecutionPlans.size() - 1).getJobSpec().getConfig().getString(ConfigurationKeys.JOB_NAME_KEY); jobConfig = jobConfig.withValue(ConfigurationKeys.JOB_DEPENDENCIES, ConfigValueFactory.fromAnyRef(previousJobName)); } FlowSpec flowSpec = FlowSpec.builder("testFlowSpec").withConfig(flowConfigs.get(i)).build(); jobExecutionPlans.add(new JobExecutionPlan.Factory().createPlan(flowSpec, jobConfig.withValue(ConfigurationKeys.JOB_TEMPLATE_PATH, ConfigValueFactory.fromAnyRef("testUri")), new InMemorySpecExecutor(ConfigFactory.empty()), 0L, ConfigFactory.empty())); } Dag<JobExecutionPlan> dag = new JobExecutionPlanDagFactory().createDag(jobExecutionPlans); Assert.assertEquals(dag.getStartNodes().size(), 1); Assert.assertEquals(dag.getEndNodes().size(), 1); Assert.assertEquals(dag.getNodes().size(), 3); Assert.assertNull(dag.getNodes().get(0).getParentNodes()); Assert.assertEquals(dag.getNodes().get(1).getParentNodes().size(), 1); Assert.assertEquals(dag.getNodes().get(2).getParentNodes().size(), 1); Assert.assertEquals(dag.getNodes().get(1).getParentNodes().get(0), dag.getNodes().get(0)); Assert.assertEquals(dag.getNodes().get(2).getParentNodes().get(0), dag.getNodes().get(1)); }
public Dag<JobExecutionPlan> createDag(List<JobExecutionPlan> jobExecutionPlans) { Map<String, Dag.DagNode<JobExecutionPlan>> jobExecutionPlanMap = new HashMap<>(); List<Dag.DagNode<JobExecutionPlan>> dagNodeList = new ArrayList<>(); for (JobExecutionPlan jobExecutionPlan : jobExecutionPlans) { Dag.DagNode<JobExecutionPlan> dagNode = new Dag.DagNode<>(jobExecutionPlan); dagNodeList.add(dagNode); String jobName = getJobName(jobExecutionPlan); if (jobName != null) { jobExecutionPlanMap.put(jobName, dagNode); } } for (JobExecutionPlan jobExecutionPlan : jobExecutionPlans) { String jobName = getJobName(jobExecutionPlan); if (jobName == null) { continue; } Dag.DagNode<JobExecutionPlan> node = jobExecutionPlanMap.get(jobName); Collection<String> dependencies = getDependencies(jobExecutionPlan.getJobSpec().getConfig()); for (String dependency : dependencies) { Dag.DagNode<JobExecutionPlan> parentNode = jobExecutionPlanMap.get(dependency); node.addParentNode(parentNode); } } Dag<JobExecutionPlan> dag = new Dag<>(dagNodeList); return dag; }
JobExecutionPlanDagFactory { public Dag<JobExecutionPlan> createDag(List<JobExecutionPlan> jobExecutionPlans) { Map<String, Dag.DagNode<JobExecutionPlan>> jobExecutionPlanMap = new HashMap<>(); List<Dag.DagNode<JobExecutionPlan>> dagNodeList = new ArrayList<>(); for (JobExecutionPlan jobExecutionPlan : jobExecutionPlans) { Dag.DagNode<JobExecutionPlan> dagNode = new Dag.DagNode<>(jobExecutionPlan); dagNodeList.add(dagNode); String jobName = getJobName(jobExecutionPlan); if (jobName != null) { jobExecutionPlanMap.put(jobName, dagNode); } } for (JobExecutionPlan jobExecutionPlan : jobExecutionPlans) { String jobName = getJobName(jobExecutionPlan); if (jobName == null) { continue; } Dag.DagNode<JobExecutionPlan> node = jobExecutionPlanMap.get(jobName); Collection<String> dependencies = getDependencies(jobExecutionPlan.getJobSpec().getConfig()); for (String dependency : dependencies) { Dag.DagNode<JobExecutionPlan> parentNode = jobExecutionPlanMap.get(dependency); node.addParentNode(parentNode); } } Dag<JobExecutionPlan> dag = new Dag<>(dagNodeList); return dag; } }
JobExecutionPlanDagFactory { public Dag<JobExecutionPlan> createDag(List<JobExecutionPlan> jobExecutionPlans) { Map<String, Dag.DagNode<JobExecutionPlan>> jobExecutionPlanMap = new HashMap<>(); List<Dag.DagNode<JobExecutionPlan>> dagNodeList = new ArrayList<>(); for (JobExecutionPlan jobExecutionPlan : jobExecutionPlans) { Dag.DagNode<JobExecutionPlan> dagNode = new Dag.DagNode<>(jobExecutionPlan); dagNodeList.add(dagNode); String jobName = getJobName(jobExecutionPlan); if (jobName != null) { jobExecutionPlanMap.put(jobName, dagNode); } } for (JobExecutionPlan jobExecutionPlan : jobExecutionPlans) { String jobName = getJobName(jobExecutionPlan); if (jobName == null) { continue; } Dag.DagNode<JobExecutionPlan> node = jobExecutionPlanMap.get(jobName); Collection<String> dependencies = getDependencies(jobExecutionPlan.getJobSpec().getConfig()); for (String dependency : dependencies) { Dag.DagNode<JobExecutionPlan> parentNode = jobExecutionPlanMap.get(dependency); node.addParentNode(parentNode); } } Dag<JobExecutionPlan> dag = new Dag<>(dagNodeList); return dag; }  }
JobExecutionPlanDagFactory { public Dag<JobExecutionPlan> createDag(List<JobExecutionPlan> jobExecutionPlans) { Map<String, Dag.DagNode<JobExecutionPlan>> jobExecutionPlanMap = new HashMap<>(); List<Dag.DagNode<JobExecutionPlan>> dagNodeList = new ArrayList<>(); for (JobExecutionPlan jobExecutionPlan : jobExecutionPlans) { Dag.DagNode<JobExecutionPlan> dagNode = new Dag.DagNode<>(jobExecutionPlan); dagNodeList.add(dagNode); String jobName = getJobName(jobExecutionPlan); if (jobName != null) { jobExecutionPlanMap.put(jobName, dagNode); } } for (JobExecutionPlan jobExecutionPlan : jobExecutionPlans) { String jobName = getJobName(jobExecutionPlan); if (jobName == null) { continue; } Dag.DagNode<JobExecutionPlan> node = jobExecutionPlanMap.get(jobName); Collection<String> dependencies = getDependencies(jobExecutionPlan.getJobSpec().getConfig()); for (String dependency : dependencies) { Dag.DagNode<JobExecutionPlan> parentNode = jobExecutionPlanMap.get(dependency); node.addParentNode(parentNode); } } Dag<JobExecutionPlan> dag = new Dag<>(dagNodeList); return dag; }  Dag<JobExecutionPlan> createDag(List<JobExecutionPlan> jobExecutionPlans); }
JobExecutionPlanDagFactory { public Dag<JobExecutionPlan> createDag(List<JobExecutionPlan> jobExecutionPlans) { Map<String, Dag.DagNode<JobExecutionPlan>> jobExecutionPlanMap = new HashMap<>(); List<Dag.DagNode<JobExecutionPlan>> dagNodeList = new ArrayList<>(); for (JobExecutionPlan jobExecutionPlan : jobExecutionPlans) { Dag.DagNode<JobExecutionPlan> dagNode = new Dag.DagNode<>(jobExecutionPlan); dagNodeList.add(dagNode); String jobName = getJobName(jobExecutionPlan); if (jobName != null) { jobExecutionPlanMap.put(jobName, dagNode); } } for (JobExecutionPlan jobExecutionPlan : jobExecutionPlans) { String jobName = getJobName(jobExecutionPlan); if (jobName == null) { continue; } Dag.DagNode<JobExecutionPlan> node = jobExecutionPlanMap.get(jobName); Collection<String> dependencies = getDependencies(jobExecutionPlan.getJobSpec().getConfig()); for (String dependency : dependencies) { Dag.DagNode<JobExecutionPlan> parentNode = jobExecutionPlanMap.get(dependency); node.addParentNode(parentNode); } } Dag<JobExecutionPlan> dag = new Dag<>(dagNodeList); return dag; }  Dag<JobExecutionPlan> createDag(List<JobExecutionPlan> jobExecutionPlans);  }