@Test (dependsOnMethods = "testCompileFlowWithRetention") public void testCompileFlowAfterFirstEdgeDeletion() throws URISyntaxException, IOException { this.flowGraph.deleteFlowEdge("HDFS-1_HDFS-1_hdfsConvertToJsonAndEncrypt"); FlowSpec spec = createFlowSpec("flow/flow1.conf", "LocalFS-1", "ADLS-1", false, false); Dag<JobExecutionPlan> jobDag = this.specCompiler.compileFlow(spec); Assert.assertEquals(jobDag.getNodes().size(), 4); Assert.assertEquals(jobDag.getStartNodes().size(), 1); Assert.assertEquals(jobDag.getEndNodes().size(), 1); DagNode<JobExecutionPlan> startNode = jobDag.getStartNodes().get(0); JobExecutionPlan jobExecutionPlan = startNode.getValue(); JobSpec jobSpec = jobExecutionPlan.getJobSpec(); Config jobConfig = jobSpec.getConfig(); String flowGroup = "testFlowGroup"; String flowName = "testFlowName"; String expectedJobName1 = Joiner.on(JobExecutionPlan.Factory.JOB_NAME_COMPONENT_SEPARATION_CHAR). join(flowGroup, flowName, "Distcp", "LocalFS-1", "HDFS-2", "localToHdfs"); String jobName1 = jobConfig.getString(ConfigurationKeys.JOB_NAME_KEY); Assert.assertTrue(jobName1.startsWith(expectedJobName1)); String from = jobConfig.getString("from"); String to = jobConfig.getString("to"); Assert.assertEquals(from, "/data/out/testTeam/testDataset"); Assert.assertEquals(to, "/data/out/testTeam/testDataset"); String sourceFsUri = jobConfig.getString("fs.uri"); Assert.assertEquals(sourceFsUri, "file: Assert.assertEquals(jobConfig.getString("source.filebased.fs.uri"), sourceFsUri); Assert.assertEquals(jobConfig.getString("state.store.fs.uri"), sourceFsUri); String targetFsUri = jobConfig.getString("target.filebased.fs.uri"); Assert.assertEquals(targetFsUri, "hdfs: Assert.assertEquals(jobConfig.getString("writer.fs.uri"), targetFsUri); Assert.assertEquals(new Path(jobConfig.getString("gobblin.dataset.pattern")), new Path(from)); Assert.assertEquals(jobConfig.getString("data.publisher.final.dir"), to); Assert.assertEquals(jobConfig.getString("type"), "java"); Assert.assertEquals(jobConfig.getString("job.class"), "org.apache.gobblin.runtime.local.LocalJobLauncher"); Assert.assertEquals(jobConfig.getString("launcher.type"), "LOCAL"); SpecExecutor specExecutor = jobExecutionPlan.getSpecExecutor(); Assert.assertEquals(specExecutor.getUri().toString(), "fs: Assert.assertEquals(specExecutor.getClass().getCanonicalName(), "org.apache.gobblin.runtime.spec_executorInstance.InMemorySpecExecutor"); Assert.assertEquals(jobDag.getChildren(startNode).size(), 1); DagNode<JobExecutionPlan> secondHopNode = jobDag.getChildren(startNode).get(0); jobExecutionPlan = secondHopNode.getValue(); jobConfig = jobExecutionPlan.getJobSpec().getConfig(); String expectedJobName2 = Joiner.on(JobExecutionPlan.Factory.JOB_NAME_COMPONENT_SEPARATION_CHAR). join(flowGroup, flowName, "ConvertToJsonAndEncrypt", "HDFS-2", "HDFS-2", "hdfsConvertToJsonAndEncrypt"); String jobName2 = jobConfig.getString(ConfigurationKeys.JOB_NAME_KEY); Assert.assertTrue(jobName2.startsWith(expectedJobName2)); Assert.assertEquals(jobConfig.getString(ConfigurationKeys.JOB_DEPENDENCIES), jobName1); from = jobConfig.getString("from"); to = jobConfig.getString("to"); Assert.assertEquals(from, "/data/out/testTeam/testDataset"); Assert.assertEquals(to, "/data/encrypted/testTeam/testDataset"); Assert.assertEquals(jobConfig.getString("source.filebased.data.directory"), from); Assert.assertEquals(jobConfig.getString("data.publisher.final.dir"), to); specExecutor = jobExecutionPlan.getSpecExecutor(); Assert.assertEquals(specExecutor.getUri().toString(), "https: Assert.assertEquals(specExecutor.getClass().getCanonicalName(), "org.apache.gobblin.service.modules.flow.MultiHopFlowCompilerTest.TestAzkabanSpecExecutor"); Assert.assertEquals(jobDag.getChildren(secondHopNode).size(), 1); DagNode<JobExecutionPlan> thirdHopNode = jobDag.getChildren(secondHopNode).get(0); jobExecutionPlan = thirdHopNode.getValue(); jobConfig = jobExecutionPlan.getJobSpec().getConfig(); String expectedJobName3 = Joiner.on(JobExecutionPlan.Factory.JOB_NAME_COMPONENT_SEPARATION_CHAR). join(flowGroup, flowName, "Distcp", "HDFS-2", "HDFS-4", "hdfsToHdfs"); String jobName3 = jobConfig.getString(ConfigurationKeys.JOB_NAME_KEY); Assert.assertTrue(jobName3.startsWith(expectedJobName3)); Assert.assertEquals(jobConfig.getString(ConfigurationKeys.JOB_DEPENDENCIES), jobName2); from = jobConfig.getString("from"); to = jobConfig.getString("to"); Assert.assertEquals(from, "/data/encrypted/testTeam/testDataset"); Assert.assertEquals(to, "/data/encrypted/testTeam/testDataset"); Assert.assertEquals(jobConfig.getString("source.filebased.fs.uri"), "hdfs: Assert.assertEquals(jobConfig.getString("target.filebased.fs.uri"), "hdfs: Assert.assertEquals(jobConfig.getString("type"), "hadoopJava"); Assert.assertEquals(jobConfig.getString("job.class"), "org.apache.gobblin.azkaban.AzkabanJobLauncher"); Assert.assertEquals(jobConfig.getString("launcher.type"), "MAPREDUCE"); specExecutor = jobExecutionPlan.getSpecExecutor(); Assert.assertEquals(specExecutor.getUri().toString(), "https: Assert.assertEquals(specExecutor.getClass().getCanonicalName(), "org.apache.gobblin.service.modules.flow.MultiHopFlowCompilerTest.TestAzkabanSpecExecutor"); Assert.assertEquals(jobDag.getChildren(thirdHopNode).size(), 1); DagNode<JobExecutionPlan> fourthHopNode = jobDag.getChildren(thirdHopNode).get(0); jobExecutionPlan = fourthHopNode.getValue(); jobConfig = jobExecutionPlan.getJobSpec().getConfig(); String expectedJobName4 = Joiner.on(JobExecutionPlan.Factory.JOB_NAME_COMPONENT_SEPARATION_CHAR). join(flowGroup, flowName, "DistcpToADL", "HDFS-4", "ADLS-1", "hdfsToAdl"); String jobName4 = jobConfig.getString(ConfigurationKeys.JOB_NAME_KEY); Assert.assertTrue(jobName4.startsWith(expectedJobName4)); Assert.assertEquals(jobConfig.getString(ConfigurationKeys.JOB_DEPENDENCIES), jobName3); from = jobConfig.getString("from"); to = jobConfig.getString("to"); Assert.assertEquals(from, "/data/encrypted/testTeam/testDataset"); Assert.assertEquals(to, "/data/encrypted/testTeam/testDataset"); Assert.assertEquals(jobConfig.getString("source.filebased.fs.uri"), "hdfs: Assert.assertEquals(jobConfig.getString("target.filebased.fs.uri"), "adl: Assert.assertEquals(jobConfig.getString("type"), "hadoopJava"); Assert.assertEquals(jobConfig.getString("job.class"), "org.apache.gobblin.azkaban.AzkabanJobLauncher"); Assert.assertEquals(jobConfig.getString("launcher.type"), "MAPREDUCE"); Assert.assertEquals(jobConfig.getString("dfs.adls.oauth2.client.id"), "1234"); Assert.assertEquals(jobConfig.getString("writer.encrypted.dfs.adls.oauth2.credential"), "credential"); Assert.assertEquals(jobConfig.getString("encrypt.key.loc"), "/user/testUser/master.password"); specExecutor = jobExecutionPlan.getSpecExecutor(); Assert.assertEquals(specExecutor.getUri().toString(), "https: Assert.assertEquals(specExecutor.getClass().getCanonicalName(), "org.apache.gobblin.service.modules.flow.MultiHopFlowCompilerTest.TestAzkabanSpecExecutor"); Assert.assertEquals(jobDag.getEndNodes().get(0), fourthHopNode); }
@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);  }