Prevent NPE in FlowCompilationValidationHelper.validateAndHandleConcurrentExecution

gobblin-service/src/main/java/org/apache/gobblin/service/modules/flow/MultiHopFlowCompiler.java

@@ -202,10 +202,10 @@ public void awaitHealthy() throws InterruptedException {
   }
 
   /**
-   * j
    * @param spec an instance of {@link FlowSpec}.
    * @return A DAG of {@link JobExecutionPlan}s, which encapsulates the compiled {@link org.apache.gobblin.runtime.api.JobSpec}s
-   * together with the {@link SpecExecutor} where the job can be executed.
+   * together with the {@link SpecExecutor} where the job can be executed; when compilation fails, return `null`, and also add a
+   * {@link org.apache.gobblin.runtime.api.FlowSpec.CompilationError} to `spec` (after casting to a {@link FlowSpec})
    */
   @Override
   public Dag<JobExecutionPlan> compileFlow(Spec spec) {

gobblin-service/src/main/java/org/apache/gobblin/service/modules/utils/FlowCompilationValidationHelper.java

@@ -115,7 +115,7 @@ public Optional<Dag<JobExecutionPlan>> validateAndHandleConcurrentExecution(Conf
     Dag<JobExecutionPlan> jobExecutionPlanDag = specCompiler.compileFlow(spec);
 
     if (isExecutionPermitted(flowStatusGenerator, flowName, flowGroup, allowConcurrentExecution)) {
-      return Optional.of(jobExecutionPlanDag);
+      return Optional.fromNullable(jobExecutionPlanDag);
     } else {
       log.warn("Another instance of flowGroup: {}, flowName: {} running; Skipping flow execution since "
           + "concurrent executions are disabled for this flow.", flowGroup, flowName);