This is an automated email from the ASF dual-hosted git repository.
arjun4084346 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/gobblin.git
The following commit(s) were added to refs/heads/master by this push:
new 6b6b70ce7 [GOBBLIN-1948] Use same flowExecutionId across participants
(#3819)
6b6b70ce7 is described below
commit 6b6b70ce7b6a2d23b9b2903bbc566e5189c7968b
Author: umustafi <[email protected]>
AuthorDate: Fri Nov 3 10:51:05 2023 -0700
[GOBBLIN-1948] Use same flowExecutionId across participants (#3819)
* Use same flowExecutionId across participants
* Set config field as well in new FlowSpec
* Use gobblin util to create config
* Rename function and move to util
---------
Co-authored-by: Urmi Mustafi <[email protected]>
---
.../org/apache/gobblin/runtime/api/FlowSpec.java | 33 ++++++-----
.../apache/gobblin/runtime/api/FlowSpecTest.java | 68 ++++++++++++++++++++++
.../monitoring/DagActionStoreChangeMonitor.java | 24 ++++----
3 files changed, 101 insertions(+), 24 deletions(-)
diff --git
a/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/api/FlowSpec.java
b/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/api/FlowSpec.java
index bb9b6de3f..2df547596 100644
--- a/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/api/FlowSpec.java
+++ b/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.google.common.collect.Sets;
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 class FlowSpec implements Configurable, Spec {
+ 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();
+ }
}
diff --git
a/gobblin-runtime/src/test/java/org/apache/gobblin/runtime/api/FlowSpecTest.java
b/gobblin-runtime/src/test/java/org/apache/gobblin/runtime/api/FlowSpecTest.java
new file mode 100644
index 000000000..793abd222
--- /dev/null
+++
b/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");
+ }
+}
diff --git
a/gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/DagActionStoreChangeMonitor.java
b/gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/DagActionStoreChangeMonitor.java
index 1190a1e46..c8d9c62b7 100644
---
a/gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/DagActionStoreChangeMonitor.java
+++
b/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.FlowId;
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 @@ public class DagActionStoreChangeMonitor extends
HighLevelConsumer {
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 @@ public class DagActionStoreChangeMonitor extends
HighLevelConsumer {
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();
