@Test (dependsOnMethods = "testCompileMultiDatasetFlow") public void testCompileCombinedDatasetFlow() throws Exception { FlowSpec spec = createFlowSpec("flow/flow4.conf", "HDFS-1", "HDFS-3", true, false); Dag<JobExecutionPlan> dag = specCompiler.compileFlow(spec); Assert.assertEquals(dag.getNodes().size(), 2); Assert.assertEquals(dag.getEndNodes().size(), 1); Assert.assertEquals(dag.getStartNodes().size(), 1); String copyJobName = Joiner.on(JobExecutionPlan.Factory.JOB_NAME_COMPONENT_SEPARATION_CHAR). join("testFlowGroup", "testFlowName", "Distcp", "HDFS-1", "HDFS-3", "hdfsToHdfs"); Config jobConfig = dag.getStartNodes().get(0).getValue().getJobSpec().getConfig(); String jobName = jobConfig.getString(ConfigurationKeys.JOB_NAME_KEY); Assert.assertTrue(jobName.startsWith(copyJobName)); Assert.assertTrue(jobConfig.getString(ConfigurableGlobDatasetFinder.DATASET_FINDER_PATTERN_KEY).endsWith("{dataset0,dataset1,dataset2}")); String retentionJobName = Joiner.on(JobExecutionPlan.Factory.JOB_NAME_COMPONENT_SEPARATION_CHAR). join("testFlowGroup", "testFlowName", "SnapshotRetention", "HDFS-3", "HDFS-3", "hdfsRetention"); Config jobConfig2 = dag.getEndNodes().get(0).getValue().getJobSpec().getConfig(); String jobName2 = jobConfig2.getString(ConfigurationKeys.JOB_NAME_KEY); Assert.assertTrue(jobName2.startsWith(retentionJobName)); Assert.assertTrue(jobConfig2.getString(ConfigurableGlobDatasetFinder.DATASET_FINDER_PATTERN_KEY).endsWith("{dataset0,dataset1,dataset2}")); }
@Override public Dag<JobExecutionPlan> compileFlow(Spec spec) { Preconditions.checkNotNull(spec); Preconditions.checkArgument(spec instanceof FlowSpec, "MultiHopFlowCompiler only accepts FlowSpecs"); long startTime = System.nanoTime(); FlowSpec flowSpec = (FlowSpec) spec; String source = ConfigUtils.getString(flowSpec.getConfig(), ServiceConfigKeys.FLOW_SOURCE_IDENTIFIER_KEY, ""); String destination = ConfigUtils.getString(flowSpec.getConfig(), ServiceConfigKeys.FLOW_DESTINATION_IDENTIFIER_KEY, ""); DataNode sourceNode = this.flowGraph.getNode(source); List<String> destNodeIds = ConfigUtils.getStringList(flowSpec.getConfig(), ServiceConfigKeys.FLOW_DESTINATION_IDENTIFIER_KEY); List<DataNode> destNodes = destNodeIds.stream().map(this.flowGraph::getNode).collect(Collectors.toList()); log.info(String.format("Compiling flow for source: %s and destination: %s", source, destination)); List<FlowSpec> flowSpecs = splitFlowSpec(flowSpec); Dag<JobExecutionPlan> jobExecutionPlanDag = new Dag<>(new ArrayList<>()); try { this.rwLock.readLock().lock(); for (FlowSpec datasetFlowSpec : flowSpecs) { for (DataNode destNode : destNodes) { long authStartTime = System.nanoTime(); boolean authorized = this.dataMovementAuthorizer.isMovementAuthorized(flowSpec, sourceNode, destNode); Instrumented.updateTimer(dataAuthorizationTimer, System.nanoTime() - authStartTime, TimeUnit.NANOSECONDS); if (!authorized) { String message = String.format("Data movement is not authorized for flow: %s, source: %s, destination: %s", flowSpec.getUri().toString(), source, destination); log.error(message); datasetFlowSpec.getCompilationErrors().add(message); return null; } } FlowGraphPath flowGraphPath = flowGraph.findPath(datasetFlowSpec); if (flowGraphPath != null) { jobExecutionPlanDag = jobExecutionPlanDag.merge(flowGraphPath.asDag(this.config)); } } if (jobExecutionPlanDag.isEmpty()) { Instrumented.markMeter(flowCompilationFailedMeter); log.info(String.format("No path found from source: %s and destination: %s", source, destination)); return jobExecutionPlanDag; } } catch (PathFinder.PathFinderException | SpecNotFoundException | JobTemplate.TemplateException | URISyntaxException | ReflectiveOperationException e) { Instrumented.markMeter(flowCompilationFailedMeter); log.error(String .format("Exception encountered while compiling flow for source: %s and destination: %s", source, destination), e); return null; } finally { this.rwLock.readLock().unlock(); } Instrumented.markMeter(flowCompilationSuccessFulMeter); Instrumented.updateTimer(flowCompilationTimer, System.nanoTime() - startTime, TimeUnit.NANOSECONDS); return jobExecutionPlanDag; }
MultiHopFlowCompiler extends BaseFlowToJobSpecCompiler { @Override public Dag<JobExecutionPlan> compileFlow(Spec spec) { Preconditions.checkNotNull(spec); Preconditions.checkArgument(spec instanceof FlowSpec, "MultiHopFlowCompiler only accepts FlowSpecs"); long startTime = System.nanoTime(); FlowSpec flowSpec = (FlowSpec) spec; String source = ConfigUtils.getString(flowSpec.getConfig(), ServiceConfigKeys.FLOW_SOURCE_IDENTIFIER_KEY, ""); String destination = ConfigUtils.getString(flowSpec.getConfig(), ServiceConfigKeys.FLOW_DESTINATION_IDENTIFIER_KEY, ""); DataNode sourceNode = this.flowGraph.getNode(source); List<String> destNodeIds = ConfigUtils.getStringList(flowSpec.getConfig(), ServiceConfigKeys.FLOW_DESTINATION_IDENTIFIER_KEY); List<DataNode> destNodes = destNodeIds.stream().map(this.flowGraph::getNode).collect(Collectors.toList()); log.info(String.format("Compiling flow for source: %s and destination: %s", source, destination)); List<FlowSpec> flowSpecs = splitFlowSpec(flowSpec); Dag<JobExecutionPlan> jobExecutionPlanDag = new Dag<>(new ArrayList<>()); try { this.rwLock.readLock().lock(); for (FlowSpec datasetFlowSpec : flowSpecs) { for (DataNode destNode : destNodes) { long authStartTime = System.nanoTime(); boolean authorized = this.dataMovementAuthorizer.isMovementAuthorized(flowSpec, sourceNode, destNode); Instrumented.updateTimer(dataAuthorizationTimer, System.nanoTime() - authStartTime, TimeUnit.NANOSECONDS); if (!authorized) { String message = String.format("Data movement is not authorized for flow: %s, source: %s, destination: %s", flowSpec.getUri().toString(), source, destination); log.error(message); datasetFlowSpec.getCompilationErrors().add(message); return null; } } FlowGraphPath flowGraphPath = flowGraph.findPath(datasetFlowSpec); if (flowGraphPath != null) { jobExecutionPlanDag = jobExecutionPlanDag.merge(flowGraphPath.asDag(this.config)); } } if (jobExecutionPlanDag.isEmpty()) { Instrumented.markMeter(flowCompilationFailedMeter); log.info(String.format("No path found from source: %s and destination: %s", source, destination)); return jobExecutionPlanDag; } } catch (PathFinder.PathFinderException | SpecNotFoundException | JobTemplate.TemplateException | URISyntaxException | ReflectiveOperationException e) { Instrumented.markMeter(flowCompilationFailedMeter); log.error(String .format("Exception encountered while compiling flow for source: %s and destination: %s", source, destination), e); return null; } finally { this.rwLock.readLock().unlock(); } Instrumented.markMeter(flowCompilationSuccessFulMeter); Instrumented.updateTimer(flowCompilationTimer, System.nanoTime() - startTime, TimeUnit.NANOSECONDS); return jobExecutionPlanDag; } }
MultiHopFlowCompiler extends BaseFlowToJobSpecCompiler { @Override public Dag<JobExecutionPlan> compileFlow(Spec spec) { Preconditions.checkNotNull(spec); Preconditions.checkArgument(spec instanceof FlowSpec, "MultiHopFlowCompiler only accepts FlowSpecs"); long startTime = System.nanoTime(); FlowSpec flowSpec = (FlowSpec) spec; String source = ConfigUtils.getString(flowSpec.getConfig(), ServiceConfigKeys.FLOW_SOURCE_IDENTIFIER_KEY, ""); String destination = ConfigUtils.getString(flowSpec.getConfig(), ServiceConfigKeys.FLOW_DESTINATION_IDENTIFIER_KEY, ""); DataNode sourceNode = this.flowGraph.getNode(source); List<String> destNodeIds = ConfigUtils.getStringList(flowSpec.getConfig(), ServiceConfigKeys.FLOW_DESTINATION_IDENTIFIER_KEY); List<DataNode> destNodes = destNodeIds.stream().map(this.flowGraph::getNode).collect(Collectors.toList()); log.info(String.format("Compiling flow for source: %s and destination: %s", source, destination)); List<FlowSpec> flowSpecs = splitFlowSpec(flowSpec); Dag<JobExecutionPlan> jobExecutionPlanDag = new Dag<>(new ArrayList<>()); try { this.rwLock.readLock().lock(); for (FlowSpec datasetFlowSpec : flowSpecs) { for (DataNode destNode : destNodes) { long authStartTime = System.nanoTime(); boolean authorized = this.dataMovementAuthorizer.isMovementAuthorized(flowSpec, sourceNode, destNode); Instrumented.updateTimer(dataAuthorizationTimer, System.nanoTime() - authStartTime, TimeUnit.NANOSECONDS); if (!authorized) { String message = String.format("Data movement is not authorized for flow: %s, source: %s, destination: %s", flowSpec.getUri().toString(), source, destination); log.error(message); datasetFlowSpec.getCompilationErrors().add(message); return null; } } FlowGraphPath flowGraphPath = flowGraph.findPath(datasetFlowSpec); if (flowGraphPath != null) { jobExecutionPlanDag = jobExecutionPlanDag.merge(flowGraphPath.asDag(this.config)); } } if (jobExecutionPlanDag.isEmpty()) { Instrumented.markMeter(flowCompilationFailedMeter); log.info(String.format("No path found from source: %s and destination: %s", source, destination)); return jobExecutionPlanDag; } } catch (PathFinder.PathFinderException | SpecNotFoundException | JobTemplate.TemplateException | URISyntaxException | ReflectiveOperationException e) { Instrumented.markMeter(flowCompilationFailedMeter); log.error(String .format("Exception encountered while compiling flow for source: %s and destination: %s", source, destination), e); return null; } finally { this.rwLock.readLock().unlock(); } Instrumented.markMeter(flowCompilationSuccessFulMeter); Instrumented.updateTimer(flowCompilationTimer, System.nanoTime() - startTime, TimeUnit.NANOSECONDS); return jobExecutionPlanDag; } MultiHopFlowCompiler(Config config);  MultiHopFlowCompiler(Config config, boolean instrumentationEnabled);  MultiHopFlowCompiler(Config config, Optional<Logger> log);  MultiHopFlowCompiler(Config config, Optional<Logger> log, boolean instrumentationEnabled); @VisibleForTesting  MultiHopFlowCompiler(Config config, FlowGraph flowGraph); }
MultiHopFlowCompiler extends BaseFlowToJobSpecCompiler { @Override public Dag<JobExecutionPlan> compileFlow(Spec spec) { Preconditions.checkNotNull(spec); Preconditions.checkArgument(spec instanceof FlowSpec, "MultiHopFlowCompiler only accepts FlowSpecs"); long startTime = System.nanoTime(); FlowSpec flowSpec = (FlowSpec) spec; String source = ConfigUtils.getString(flowSpec.getConfig(), ServiceConfigKeys.FLOW_SOURCE_IDENTIFIER_KEY, ""); String destination = ConfigUtils.getString(flowSpec.getConfig(), ServiceConfigKeys.FLOW_DESTINATION_IDENTIFIER_KEY, ""); DataNode sourceNode = this.flowGraph.getNode(source); List<String> destNodeIds = ConfigUtils.getStringList(flowSpec.getConfig(), ServiceConfigKeys.FLOW_DESTINATION_IDENTIFIER_KEY); List<DataNode> destNodes = destNodeIds.stream().map(this.flowGraph::getNode).collect(Collectors.toList()); log.info(String.format("Compiling flow for source: %s and destination: %s", source, destination)); List<FlowSpec> flowSpecs = splitFlowSpec(flowSpec); Dag<JobExecutionPlan> jobExecutionPlanDag = new Dag<>(new ArrayList<>()); try { this.rwLock.readLock().lock(); for (FlowSpec datasetFlowSpec : flowSpecs) { for (DataNode destNode : destNodes) { long authStartTime = System.nanoTime(); boolean authorized = this.dataMovementAuthorizer.isMovementAuthorized(flowSpec, sourceNode, destNode); Instrumented.updateTimer(dataAuthorizationTimer, System.nanoTime() - authStartTime, TimeUnit.NANOSECONDS); if (!authorized) { String message = String.format("Data movement is not authorized for flow: %s, source: %s, destination: %s", flowSpec.getUri().toString(), source, destination); log.error(message); datasetFlowSpec.getCompilationErrors().add(message); return null; } } FlowGraphPath flowGraphPath = flowGraph.findPath(datasetFlowSpec); if (flowGraphPath != null) { jobExecutionPlanDag = jobExecutionPlanDag.merge(flowGraphPath.asDag(this.config)); } } if (jobExecutionPlanDag.isEmpty()) { Instrumented.markMeter(flowCompilationFailedMeter); log.info(String.format("No path found from source: %s and destination: %s", source, destination)); return jobExecutionPlanDag; } } catch (PathFinder.PathFinderException | SpecNotFoundException | JobTemplate.TemplateException | URISyntaxException | ReflectiveOperationException e) { Instrumented.markMeter(flowCompilationFailedMeter); log.error(String .format("Exception encountered while compiling flow for source: %s and destination: %s", source, destination), e); return null; } finally { this.rwLock.readLock().unlock(); } Instrumented.markMeter(flowCompilationSuccessFulMeter); Instrumented.updateTimer(flowCompilationTimer, System.nanoTime() - startTime, TimeUnit.NANOSECONDS); return jobExecutionPlanDag; } MultiHopFlowCompiler(Config config);  MultiHopFlowCompiler(Config config, boolean instrumentationEnabled);  MultiHopFlowCompiler(Config config, Optional<Logger> log);  MultiHopFlowCompiler(Config config, Optional<Logger> log, boolean instrumentationEnabled); @VisibleForTesting  MultiHopFlowCompiler(Config config, FlowGraph flowGraph); @Override void setActive(boolean active); @Override void awaitHealthy(); @Override Dag<JobExecutionPlan> compileFlow(Spec spec); }
MultiHopFlowCompiler extends BaseFlowToJobSpecCompiler { @Override public Dag<JobExecutionPlan> compileFlow(Spec spec) { Preconditions.checkNotNull(spec); Preconditions.checkArgument(spec instanceof FlowSpec, "MultiHopFlowCompiler only accepts FlowSpecs"); long startTime = System.nanoTime(); FlowSpec flowSpec = (FlowSpec) spec; String source = ConfigUtils.getString(flowSpec.getConfig(), ServiceConfigKeys.FLOW_SOURCE_IDENTIFIER_KEY, ""); String destination = ConfigUtils.getString(flowSpec.getConfig(), ServiceConfigKeys.FLOW_DESTINATION_IDENTIFIER_KEY, ""); DataNode sourceNode = this.flowGraph.getNode(source); List<String> destNodeIds = ConfigUtils.getStringList(flowSpec.getConfig(), ServiceConfigKeys.FLOW_DESTINATION_IDENTIFIER_KEY); List<DataNode> destNodes = destNodeIds.stream().map(this.flowGraph::getNode).collect(Collectors.toList()); log.info(String.format("Compiling flow for source: %s and destination: %s", source, destination)); List<FlowSpec> flowSpecs = splitFlowSpec(flowSpec); Dag<JobExecutionPlan> jobExecutionPlanDag = new Dag<>(new ArrayList<>()); try { this.rwLock.readLock().lock(); for (FlowSpec datasetFlowSpec : flowSpecs) { for (DataNode destNode : destNodes) { long authStartTime = System.nanoTime(); boolean authorized = this.dataMovementAuthorizer.isMovementAuthorized(flowSpec, sourceNode, destNode); Instrumented.updateTimer(dataAuthorizationTimer, System.nanoTime() - authStartTime, TimeUnit.NANOSECONDS); if (!authorized) { String message = String.format("Data movement is not authorized for flow: %s, source: %s, destination: %s", flowSpec.getUri().toString(), source, destination); log.error(message); datasetFlowSpec.getCompilationErrors().add(message); return null; } } FlowGraphPath flowGraphPath = flowGraph.findPath(datasetFlowSpec); if (flowGraphPath != null) { jobExecutionPlanDag = jobExecutionPlanDag.merge(flowGraphPath.asDag(this.config)); } } if (jobExecutionPlanDag.isEmpty()) { Instrumented.markMeter(flowCompilationFailedMeter); log.info(String.format("No path found from source: %s and destination: %s", source, destination)); return jobExecutionPlanDag; } } catch (PathFinder.PathFinderException | SpecNotFoundException | JobTemplate.TemplateException | URISyntaxException | ReflectiveOperationException e) { Instrumented.markMeter(flowCompilationFailedMeter); log.error(String .format("Exception encountered while compiling flow for source: %s and destination: %s", source, destination), e); return null; } finally { this.rwLock.readLock().unlock(); } Instrumented.markMeter(flowCompilationSuccessFulMeter); Instrumented.updateTimer(flowCompilationTimer, System.nanoTime() - startTime, TimeUnit.NANOSECONDS); return jobExecutionPlanDag; } MultiHopFlowCompiler(Config config);  MultiHopFlowCompiler(Config config, boolean instrumentationEnabled);  MultiHopFlowCompiler(Config config, Optional<Logger> log);  MultiHopFlowCompiler(Config config, Optional<Logger> log, boolean instrumentationEnabled); @VisibleForTesting  MultiHopFlowCompiler(Config config, FlowGraph flowGraph); @Override void setActive(boolean active); @Override void awaitHealthy(); @Override Dag<JobExecutionPlan> compileFlow(Spec spec);  }