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

[suvasude]

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 <jmose...@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;
       }