This is an automated email from the ASF dual-hosted git repository.
mattcasters pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hop.git
The following commit(s) were added to refs/heads/main by this push:
new 3e53c72232 fixes #7743: stop overwriting workflow parentId with Spark
variable name (#7746)
3e53c72232 is described below
commit 3e53c7223266ca3f247c5c37956f6c7a2a396f2d
Author: Matt Casters <[email protected]>
AuthorDate: Sun Aug 2 13:31:51 2026 +0200
fixes #7743: stop overwriting workflow parentId with Spark variable name
(#7746)
LocalWorkflowEngine rebind used resolve("Internal.Spark.TransformOwnerId"),
which returns the bare name unchanged when no ${} tokens are present, so
every
local workflow got that literal as PARENT_ID. Use getVariable() so rebind
only
applies when Spark actually set the owner id.
---
.../engines/local/LocalWorkflowEngine.java | 32 +++++-
.../LocalWorkflowEngineSparkParentRebindTest.java | 120 +++++++++++++++++++++
2 files changed, 148 insertions(+), 4 deletions(-)
diff --git
a/engine/src/main/java/org/apache/hop/workflow/engines/local/LocalWorkflowEngine.java
b/engine/src/main/java/org/apache/hop/workflow/engines/local/LocalWorkflowEngine.java
index 5d673aa4d7..916ebd609b 100644
---
a/engine/src/main/java/org/apache/hop/workflow/engines/local/LocalWorkflowEngine.java
+++
b/engine/src/main/java/org/apache/hop/workflow/engines/local/LocalWorkflowEngine.java
@@ -61,6 +61,12 @@ import org.apache.hop.workflow.engine.WorkflowEnginePlugin;
description = "Executes your workflow locally")
public class LocalWorkflowEngine extends Workflow implements
IWorkflowEngine<WorkflowMeta> {
+ /**
+ * Must stay in sync with {@code
org.apache.hop.spark.util.SparkConst#VAR_TRANSFORM_OWNER_ID}.
+ * Engine cannot depend on the spark plugin.
+ */
+ static final String VAR_SPARK_TRANSFORM_OWNER_ID =
"Internal.Spark.TransformOwnerId";
+
private ExecutionInfoLocation executionInfoLocation;
private Timer executionInfoTimer;
@@ -325,27 +331,45 @@ public class LocalWorkflowEngine extends Workflow
implements IWorkflowEngine<Wor
* When this workflow is nested under a Native Spark mapPartitions transform
(Workflow Executor),
* the parent transform is registered under a synthetic id {@code
pipelineId|name|copy}. Rebind so
* the execution perspective can drill down from that transform node.
+ *
+ * <p>Uses {@link IVariables#getVariable(String)} (not {@link
IVariables#resolve(String)}):
+ * resolve only substitutes {@code ${...}} tokens and returns a bare name
unchanged, which would
+ * always overwrite parentId with the literal variable name (issue #7743).
*/
- private void rebindSparkTransformOwnerParent(Execution execution) {
+ /** Package-private for unit tests. */
+ void rebindSparkTransformOwnerParent(Execution execution) {
if (execution == null) {
return;
}
- String sparkOwner = resolve("Internal.Spark.TransformOwnerId");
+ String sparkOwner = sparkTransformOwnerId(this);
if (StringUtils.isNotEmpty(sparkOwner)) {
execution.setParentId(sparkOwner);
}
}
- private void rebindSparkTransformOwnerParent(ExecutionState state) {
+ /** Package-private for unit tests. */
+ void rebindSparkTransformOwnerParent(ExecutionState state) {
if (state == null) {
return;
}
- String sparkOwner = resolve("Internal.Spark.TransformOwnerId");
+ String sparkOwner = sparkTransformOwnerId(this);
if (StringUtils.isNotEmpty(sparkOwner)) {
state.setParentId(sparkOwner);
}
}
+ /**
+ * Returns the Spark transform owner id when set on the variable space;
otherwise null.
+ *
+ * <p>Package-private for unit tests.
+ */
+ static String sparkTransformOwnerId(IVariables variables) {
+ if (variables == null) {
+ return null;
+ }
+ return variables.getVariable(VAR_SPARK_TRANSFORM_OWNER_ID);
+ }
+
public void startExecutionInfoTimer() {
if (executionInfoLocation == null) {
return;
diff --git
a/engine/src/test/java/org/apache/hop/workflow/engines/local/LocalWorkflowEngineSparkParentRebindTest.java
b/engine/src/test/java/org/apache/hop/workflow/engines/local/LocalWorkflowEngineSparkParentRebindTest.java
new file mode 100644
index 0000000000..d25001e12e
--- /dev/null
+++
b/engine/src/test/java/org/apache/hop/workflow/engines/local/LocalWorkflowEngineSparkParentRebindTest.java
@@ -0,0 +1,120 @@
+/*
+ * 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.hop.workflow.engines.local;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+import org.apache.hop.core.variables.Variables;
+import org.apache.hop.execution.Execution;
+import org.apache.hop.execution.ExecutionState;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Regression for issue #7743: parentId must not be overwritten with the
literal variable name
+ * {@code Internal.Spark.TransformOwnerId} when the Spark owner variable is
unset.
+ */
+class LocalWorkflowEngineSparkParentRebindTest {
+
+ @Test
+ void sparkTransformOwnerId_nullWhenUnset() {
+ Variables vars = new Variables();
+ vars.initializeFrom(null);
+ assertNull(LocalWorkflowEngine.sparkTransformOwnerId(vars));
+ }
+
+ @Test
+ void sparkTransformOwnerId_returnsValueWhenSet() {
+ Variables vars = new Variables();
+ vars.initializeFrom(null);
+ vars.setVariable(LocalWorkflowEngine.VAR_SPARK_TRANSFORM_OWNER_ID,
"pipe|Workflow Executor|0");
+ assertEquals("pipe|Workflow Executor|0",
LocalWorkflowEngine.sparkTransformOwnerId(vars));
+ }
+
+ @Test
+ void sparkTransformOwnerId_nullVariables() {
+ assertNull(LocalWorkflowEngine.sparkTransformOwnerId(null));
+ }
+
+ @Test
+ void rebind_doesNotOverwriteParentWhenSparkVariableUnset() {
+ LocalWorkflowEngine engine = new LocalWorkflowEngine();
+ engine.initializeFrom(null);
+
+ Execution execution = new Execution();
+ execution.setParentId("parent-workflow-uuid");
+ engine.rebindSparkTransformOwnerParent(execution);
+ assertEquals("parent-workflow-uuid", execution.getParentId());
+
+ Execution root = new Execution();
+ root.setParentId(null);
+ engine.rebindSparkTransformOwnerParent(root);
+ assertNull(root.getParentId());
+ }
+
+ @Test
+ void rebind_setsSparkOwnerWhenVariablePresent() {
+ LocalWorkflowEngine engine = new LocalWorkflowEngine();
+ engine.initializeFrom(null);
+ engine.setVariable(
+ LocalWorkflowEngine.VAR_SPARK_TRANSFORM_OWNER_ID, "spark-pipe|Workflow
Executor|0");
+
+ Execution execution = new Execution();
+ execution.setParentId("would-be-overwritten");
+ engine.rebindSparkTransformOwnerParent(execution);
+ assertEquals("spark-pipe|Workflow Executor|0", execution.getParentId());
+
+ ExecutionState state = new ExecutionState();
+ state.setParentId("state-parent");
+ engine.rebindSparkTransformOwnerParent(state);
+ assertEquals("spark-pipe|Workflow Executor|0", state.getParentId());
+ }
+
+ @Test
+ void rebind_ignoresEmptySparkOwner() {
+ LocalWorkflowEngine engine = new LocalWorkflowEngine();
+ engine.initializeFrom(null);
+ engine.setVariable(LocalWorkflowEngine.VAR_SPARK_TRANSFORM_OWNER_ID, "");
+
+ Execution execution = new Execution();
+ execution.setParentId("keep-me");
+ engine.rebindSparkTransformOwnerParent(execution);
+ assertEquals("keep-me", execution.getParentId());
+ }
+
+ /**
+ * Documents the bug: {@code resolve(bareName)} returns the name unchanged
and must not be used
+ * for variable lookup.
+ */
+ @Test
+ void resolveBareName_returnsLiteral_notVariableValue() {
+ Variables vars = new Variables();
+ vars.initializeFrom(null);
+ // Unset: resolve leaves the bare token as-is (no ${} markers)
+ assertEquals(
+ LocalWorkflowEngine.VAR_SPARK_TRANSFORM_OWNER_ID,
+ vars.resolve(LocalWorkflowEngine.VAR_SPARK_TRANSFORM_OWNER_ID));
+ // Set: resolve still returns the bare name (it is not a ${var} expression)
+ vars.setVariable(LocalWorkflowEngine.VAR_SPARK_TRANSFORM_OWNER_ID,
"actual-owner");
+ assertEquals(
+ LocalWorkflowEngine.VAR_SPARK_TRANSFORM_OWNER_ID,
+ vars.resolve(LocalWorkflowEngine.VAR_SPARK_TRANSFORM_OWNER_ID));
+ assertEquals(
+ "actual-owner",
vars.getVariable(LocalWorkflowEngine.VAR_SPARK_TRANSFORM_OWNER_ID));
+ }
+}
