This is an automated email from the ASF dual-hosted git repository.
wlo 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 290d17d65 [GOBBLIN-2075] Fix JobSpec fetching from DagManagerUtils so
that properties are properly reflecting job configs (#3956)
290d17d65 is described below
commit 290d17d6556b154ec4cbe8b99d9d2209e6d41588
Author: William Lo <[email protected]>
AuthorDate: Mon Jun 3 15:44:12 2024 -0400
[GOBBLIN-2075] Fix JobSpec fetching from DagManagerUtils so that properties
are properly reflecting job configs (#3956)
Fix bug where dagnode jobspec properties when fetching using Utils was not
actually copying properties
---
.../modules/orchestration/DagManagerUtils.java | 2 +-
.../modules/orchestration/proc/DagProcUtils.java | 3 +-
.../modules/orchestration/DagManagerUtilsTest.java | 41 ++++++++++++++++++++++
.../GaaSJobObservabilityProducerTest.java | 3 +-
4 files changed, 46 insertions(+), 3 deletions(-)
diff --git
a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagerUtils.java
b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagerUtils.java
index e2fc6a179..92d81c7be 100644
---
a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagerUtils.java
+++
b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagerUtils.java
@@ -184,7 +184,7 @@ public class DagManagerUtils {
JobSpec jobSpec = dagNode.getValue().getJobSpec();
Map<String, String> configWithCurrentAttempts =
ImmutableMap.of(ConfigurationKeys.JOB_CURRENT_ATTEMPTS,
String.valueOf(dagNode.getValue().getCurrentAttempts()),
ConfigurationKeys.JOB_CURRENT_GENERATION,
String.valueOf(dagNode.getValue().getCurrentGeneration()));
- Properties configAsProperties = new
Properties(jobSpec.getConfigAsProperties());
+ Properties configAsProperties = (Properties)
jobSpec.getConfigAsProperties().clone();
configAsProperties.putAll(configWithCurrentAttempts);
//Return new spec with new config to avoid change the reference to dagNode
return new JobSpec(jobSpec.getUri(), jobSpec.getVersion(),
jobSpec.getDescription(),
ConfigFactory.parseMap(configWithCurrentAttempts).withFallback(jobSpec.getConfig()),
diff --git
a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/DagProcUtils.java
b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/DagProcUtils.java
index 41615c79b..91bdbf68d 100644
---
a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/DagProcUtils.java
+++
b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/DagProcUtils.java
@@ -45,6 +45,7 @@ import
org.apache.gobblin.service.modules.orchestration.DagManagerUtils;
import org.apache.gobblin.service.modules.orchestration.TimingEventUtils;
import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
import org.apache.gobblin.util.ConfigUtils;
+import org.apache.gobblin.util.PropertiesUtils;
import static org.apache.gobblin.service.ExecutionStatus.CANCELLED;
@@ -100,7 +101,7 @@ public class DagProcUtils {
jobExecutionPlan.setExecutionStatus(ExecutionStatus.ORCHESTRATED);
jobMetadata.put(TimingEvent.METADATA_MESSAGE,
producer.getExecutionLink(addSpecFuture, specExecutorUri));
// Add serialized job properties as part of the orchestrated job event
metadata
- jobMetadata.put(JobExecutionPlan.JOB_PROPS_KEY,
dagNode.getValue().toString());
+ jobMetadata.put(JobExecutionPlan.JOB_PROPS_KEY,
PropertiesUtils.serialize(jobSpec.getConfigAsProperties()));
jobOrchestrationTimer.stop(jobMetadata);
log.info("Orchestrated job: {} on Executor: {}",
DagManagerUtils.getFullyQualifiedJobName(dagNode), specExecutorUri);
dagManagementStateStore.getDagManagerMetrics().incrementJobsSentToExecutor(dagNode);
diff --git
a/gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/DagManagerUtilsTest.java
b/gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/DagManagerUtilsTest.java
new file mode 100644
index 000000000..63c521749
--- /dev/null
+++
b/gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/DagManagerUtilsTest.java
@@ -0,0 +1,41 @@
+/*
+ * 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.service.modules.orchestration;
+
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import org.apache.gobblin.runtime.api.JobSpec;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+
+
+public class DagManagerUtilsTest {
+
+ @Test
+ public void testGetJobSpecFromDag() throws Exception {
+ Dag<JobExecutionPlan> testDag = DagTestUtils.buildDag("testDag", 1000L);
+ JobSpec jobSpec = DagManagerUtils.getJobSpec(testDag.getNodes().get(0));
+ Assert.assertEquals(jobSpec.getConfigAsProperties().size(),
jobSpec.getConfig().entrySet().size());
+ for (String key : jobSpec.getConfigAsProperties().stringPropertyNames()) {
+ Assert.assertTrue(jobSpec.getConfig().hasPath(key));
+ // Assume each key is a string because all job configs are currently
strings
+ Assert.assertEquals(jobSpec.getConfigAsProperties().get(key),
jobSpec.getConfig().getString(key));
+ }
+ }
+}
diff --git
a/gobblin-service/src/test/java/org/apache/gobblin/service/monitoring/GaaSJobObservabilityProducerTest.java
b/gobblin-service/src/test/java/org/apache/gobblin/service/monitoring/GaaSJobObservabilityProducerTest.java
index 1f3da5205..2aaf7ac26 100644
---
a/gobblin-service/src/test/java/org/apache/gobblin/service/monitoring/GaaSJobObservabilityProducerTest.java
+++
b/gobblin-service/src/test/java/org/apache/gobblin/service/monitoring/GaaSJobObservabilityProducerTest.java
@@ -30,6 +30,7 @@ import org.testng.Assert;
import org.testng.annotations.Test;
import com.google.common.collect.Maps;
+import com.google.gson.JsonParser;
import io.opentelemetry.api.common.AttributeKey;
import io.opentelemetry.sdk.metrics.data.LongPointData;
@@ -136,7 +137,7 @@ public class GaaSJobObservabilityProducerTest {
Assert.assertEquals(event.getDatasetsMetrics().get(1).getEntitiesWritten(),
Long.valueOf(dataset2.getRecordsWritten()));
Assert.assertEquals(event.getDatasetsMetrics().get(1).getBytesWritten(),
Long.valueOf(dataset2.getBytesWritten()));
Assert.assertEquals(event.getDatasetsMetrics().get(1).getSuccessfullyCommitted(),
Boolean.valueOf(dataset2.isSuccessfullyCommitted()));
- Assert.assertEquals(event.getJobProperties(),
"{\"gobblin.flow.sourceIdentifier\":\"sourceNode\",\"gobblin.flow.destinationIdentifier\":\"destinationNode\",\"user.to.proxy\":\"newUser\",\"flow.executionId\":\"1681242538558\"}");
+ JsonParser.parseString(event.getJobProperties()); // Should not throw
Assert.assertEquals(event.getGaasId(), "testCluster");
AvroSerializer<GaaSJobObservabilityEvent> serializer = new
AvroBinarySerializer<>(
GaaSJobObservabilityEvent.SCHEMA$, new NoopSchemaVersionWriter()
