[incubator-gobblin] branch master updated: [GOBBLIN-1262] Update flow execution as failed if it was skipped due to concurrently running flow

[su...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

suvasude pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-gobblin.git


The following commit(s) were added to refs/heads/master by this push:
     new baebccd  [GOBBLIN-1262] Update flow execution as failed if it was skipped due to concurrently running flow
baebccd is described below

commit baebccde966b48c1b3ba6092ef6491c51a578341
Author: Jack Moseley <jm...@linkedin.com>
AuthorDate: Mon Sep 14 11:53:01 2020 -0700

    [GOBBLIN-1262] Update flow execution as failed if it was skipped due to concurrently running flow
    
    Closes #3103 from jack-moseley/status-concurrent
---
 .../gobblin/service/modules/orchestration/Orchestrator.java       | 8 ++++++++
 1 file changed, 8 insertions(+)

diff --git a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java
index af7eb94..e6afe93 100644
--- a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java
+++ b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java
@@ -247,6 +247,14 @@ public class Orchestrator implements SpecCatalogListener, Instrumentable {
         _log.warn("Another instance of flowGroup: {}, flowName: {} running; Skipping flow execution since "
             + "concurrent executions are disabled for this flow.", flowGroup, flowName);
         flowGauges.get(spec.getUri().toString()).setState(CompiledState.SKIPPED);
+
+        // Send FLOW_FAILED event
+        Map<String, String> flowMetadata = TimingEventUtils.getFlowMetadata((FlowSpec) spec);
+        flowMetadata.put(TimingEvent.METADATA_MESSAGE, "Flow failed because another instance is running and concurrent "
+            + "executions are disabled. Set flow.allowConcurrentExecution to true in the flow spec to change this behaviour.");
+        if (this.eventSubmitter.isPresent()) {
+          this.eventSubmitter.get().getTimingEvent(TimingEvent.FlowTimings.FLOW_FAILED).stop(flowMetadata);
+        }
         return;
       }