[GOBBLIN-1948] Use same flowExecutionId across participants

gobblin-runtime/src/main/java/org/apache/gobblin/runtime/api/FlowSpec.java

@@ -17,17 +17,6 @@
 
 package org.apache.gobblin.runtime.api;
 
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.List;
-import java.util.Properties;
-import java.util.Set;
-
-import org.apache.commons.lang.StringUtils;
-
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
@@ -37,12 +26,20 @@
 import com.linkedin.data.template.StringMap;
 import com.typesafe.config.Config;
 import com.typesafe.config.ConfigFactory;
-
+import com.typesafe.config.ConfigValueFactory;
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
 import lombok.Data;
 import lombok.EqualsAndHashCode;
 import lombok.extern.slf4j.Slf4j;
-
+import org.apache.commons.lang.StringUtils;
 import org.apache.gobblin.annotation.Alpha;
 import org.apache.gobblin.configuration.ConfigurationKeys;
 import org.apache.gobblin.service.FlowConfig;
@@ -522,4 +519,14 @@ public static int maxFlowSpecUriLength() {
         + URI_PATH_SEPARATOR.length() + ServiceConfigKeys.MAX_FLOW_NAME_LENGTH + URI_PATH_SEPARATOR.length() + ServiceConfigKeys.MAX_FLOW_GROUP_LENGTH;
     }
   }
+
+  /**
+   * Create a new FlowSpec object with the added property defined by path and value parameters
+   * @param path key for new property
+   * @param value
+   */
+  public static FlowSpec createFlowSpecWithProperty(FlowSpec flowSpec, String path, String value) {
+    Config updatedConfig = flowSpec.getConfig().withValue(path, ConfigValueFactory.fromAnyRef(value));
+    return new Builder(flowSpec.getUri()).withConfig(updatedConfig).build();
+  }
 }

gobblin-runtime/src/test/java/org/apache/gobblin/runtime/api/FlowSpecTest.java

@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.runtime.api;
+
+import com.typesafe.config.Config;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.Properties;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.service.FlowId;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import static org.apache.gobblin.runtime.api.FlowSpec.*;
+
+
+public class FlowSpecTest {
+
+  /**
+   * Tests that the addProperty() function to ensure the new flowSpec returned has the original properties and updated
+   * ones
+   * @throws URISyntaxException
+   */
+  @Test
+  public void testAddProperty() throws URISyntaxException {
+    String flowGroup = "myGroup";
+    String flowName = "myName";
+    String flowExecutionId = "myId";
+    FlowId flowId = new FlowId().setFlowGroup(flowGroup).setFlowName(flowName);
+    URI flowUri = FlowSpec.Utils.createFlowSpecUri(flowId);
+
+    // Create properties to be used as config
+    Properties properties = new Properties();
+    properties.setProperty(ConfigurationKeys.FLOW_GROUP_KEY, flowGroup);
+    properties.setProperty(ConfigurationKeys.FLOW_NAME_KEY, flowName);
+    properties.setProperty(ConfigurationKeys.FLOW_IS_REMINDER_EVENT_KEY, "true");
+
+    FlowSpec originalFlowSpec = FlowSpec.builder(flowUri).withConfigAsProperties(properties).build();
+    FlowSpec updatedFlowSpec = createFlowSpecWithProperty(originalFlowSpec, ConfigurationKeys.FLOW_EXECUTION_ID_KEY, flowExecutionId);
+
+    Properties updatedProperties = updatedFlowSpec.getConfigAsProperties();
+    Assert.assertEquals(updatedProperties.getProperty(ConfigurationKeys.FLOW_EXECUTION_ID_KEY), flowExecutionId);
+    Assert.assertEquals(updatedProperties.getProperty(ConfigurationKeys.FLOW_GROUP_KEY), flowGroup);
+    Assert.assertEquals(updatedProperties.getProperty(ConfigurationKeys.FLOW_NAME_KEY), flowName);
+    Assert.assertEquals(updatedProperties.getProperty(ConfigurationKeys.FLOW_IS_REMINDER_EVENT_KEY), "true");
+
+    Config updatedConfig = updatedFlowSpec.getConfig();
+    Assert.assertEquals(updatedConfig.getString(ConfigurationKeys.FLOW_EXECUTION_ID_KEY), flowExecutionId);
+    Assert.assertEquals(updatedConfig.getString(ConfigurationKeys.FLOW_GROUP_KEY), flowGroup);
+    Assert.assertEquals(updatedConfig.getString(ConfigurationKeys.FLOW_NAME_KEY), flowName);
+    Assert.assertEquals(updatedConfig.getString(ConfigurationKeys.FLOW_IS_REMINDER_EVENT_KEY), "true");
+  }
+}

gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/DagActionStoreChangeMonitor.java

@@ -17,20 +17,18 @@
 
 package org.apache.gobblin.service.monitoring;
 
-import java.io.IOException;
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.util.UUID;
-import java.util.concurrent.TimeUnit;
-
 import com.google.common.cache.CacheBuilder;
 import com.google.common.cache.CacheLoader;
 import com.google.common.cache.LoadingCache;
 import com.typesafe.config.Config;
 import com.typesafe.config.ConfigValueFactory;
-
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
 import lombok.extern.slf4j.Slf4j;
-
+import org.apache.gobblin.configuration.ConfigurationKeys;
 import org.apache.gobblin.kafka.client.DecodeableKafkaRecord;
 import org.apache.gobblin.metrics.ContextAwareGauge;
 import org.apache.gobblin.metrics.ContextAwareMeter;
@@ -44,6 +42,8 @@
 import org.apache.gobblin.service.modules.orchestration.DagManager;
 import org.apache.gobblin.service.modules.orchestration.Orchestrator;
 
+import static org.apache.gobblin.runtime.api.FlowSpec.*;
+
 
 /**
  * A DagActionStore change monitor that uses {@link DagActionStoreChangeEvent} schema to process Kafka messages received
@@ -165,7 +165,7 @@ protected void processMessage(DecodeableKafkaRecord message) {
             throw new RuntimeException(String.format("Received LAUNCH dagAction while not in multi-active scheduler "
                 + "mode for flowAction: %s", dagAction));
           }
-          submitFlowToDagManagerHelper(flowGroup, flowName);
+          submitFlowToDagManagerHelper(flowGroup, flowName, flowExecutionId);
         } else {
           log.warn("Received unsupported dagAction {}. Expected to be a KILL, RESUME, or LAUNCH", dagActionType);
           this.unexpectedErrors.mark();
@@ -192,14 +192,16 @@ protected void processMessage(DecodeableKafkaRecord message) {
     dagActionsSeenCache.put(changeIdentifier, changeIdentifier);
   }
 
-  protected void submitFlowToDagManagerHelper(String flowGroup, String flowName) {
+  protected void submitFlowToDagManagerHelper(String flowGroup, String flowName, String flowExecutionId) {
     // Retrieve job execution plan by recompiling the flow spec to send to the DagManager
     FlowId flowId = new FlowId().setFlowGroup(flowGroup).setFlowName(flowName);
     FlowSpec spec = null;
     try {
       URI flowUri = FlowSpec.Utils.createFlowSpecUri(flowId);
       spec = (FlowSpec) flowCatalog.getSpecs(flowUri);
-      this.orchestrator.submitFlowToDagManager(spec);
+      // Adds flowExecutionId to config to ensure they are consistent across hosts
+      FlowSpec updatedSpec = createFlowSpecWithProperty(spec, ConfigurationKeys.FLOW_EXECUTION_ID_KEY, flowExecutionId);
+      this.orchestrator.submitFlowToDagManager(updatedSpec);
     } catch (URISyntaxException e) {
       log.warn("Could not create URI object for flowId {}. Exception {}", flowId, e.getMessage());
       this.failedFlowLaunchSubmissions.mark();