mattcasters commented on code in PR #9059:
URL: https://github.com/apache/hop/pull/9059#discussion_r4336094153
##########
plugins/transforms/pipelineexecutor/src/main/java/org/apache/hop/pipeline/transforms/pipelineexecutor/PipelineExecutor.java:
##########
@@ -330,6 +339,87 @@ Result executeWithRetries(List<String> parameterValues)
throws HopException {
return result;
}
+ @VisibleForTesting
+ void refreshRetriedParameterVariablesFromExternalScopes() {
+ if (!meta.isInheritingAllVariables()) {
+ return;
+ }
+
+ List<PipelineExecutorParameters> parameters = meta.getParameters();
+ if (parameters == null) {
+ return;
+ }
+
+ for (PipelineExecutorParameters parameter : parameters) {
+ if (parameter == null) {
+ continue;
+ }
+
+ // Explicit field/static mappings remain authoritative.
+ if (!Utils.isEmpty(Const.trim(parameter.getField()))
+ || !Utils.isEmpty(Const.trim(parameter.getInput()))) {
+ continue;
+ }
+
+ String variableName = Const.trim(parameter.getVariable());
+ if (Utils.isEmpty(variableName)) {
+ continue;
+ }
+
+ String latestValue =
resolveRetriedVariableValueFromExternalScopes(variableName);
+ if (latestValue != null) {
+ setVariable(variableName, latestValue);
+ }
+ }
+ }
+
+ private String resolveRetriedVariableValueFromExternalScopes(String
variableName) {
+ IPipelineEngine<PipelineMeta> pipeline = getPipeline();
+ if (pipeline != null) {
+ String pipelineValue = pipeline.getVariable(variableName);
+ if (pipelineValue != null) {
+ return pipelineValue;
+ }
+ }
+
+ IVariables parent = pipeline != null ? getParentEngine(pipeline) : null;
+ while (parent != null) {
+ String value = parent.getVariable(variableName);
+ if (value != null) {
+ return value;
+ }
+ parent = getParentEngine(parent);
+ }
+
+ return System.getProperty(variableName);
+ }
+
+ private IVariables getParentEngine(IVariables engine) {
+ if (engine instanceof IPipelineEngine) {
+ IPipelineEngine<?> parentPipeline = (IPipelineEngine<?>) engine;
+ if (parentPipeline.getParentPipeline() != null) {
+ return parentPipeline.getParentPipeline();
+ }
+ return parentPipeline.getParentWorkflow();
+ }
+ if (engine instanceof IWorkflowEngine) {
+ IWorkflowEngine<?> parentWorkflow = (IWorkflowEngine<?>) engine;
+ if (parentWorkflow.getParentPipeline() != null) {
+ return parentWorkflow.getParentPipeline();
+ }
+ return parentWorkflow.getParentWorkflow();
+ }
+ return null;
+ }
+
+ @VisibleForTesting
+ List<String> resolveParameterValuesForRetryAttempt(List<String>
previousAttemptParameterValues) {
+ List<String> latestIncomingFieldValues = getLastIncomingFieldValues();
+ return !Utils.isEmpty(latestIncomingFieldValues)
+ ? latestIncomingFieldValues
Review Comment:
**[bug]** Every retry discards the parameter list used for attempt 0 and
substitutes `getLastIncomingFieldValues()` whenever the group buffer's last row
has a non-null cell. That buffer is not updated while `executeWithRetries`
runs, so this is not a newer row. It is a different encoding of whatever is
already buffered, and it is applied even when "inherit all variables" is off.
`processRow` builds the original list with `IValueMeta.getString` and keeps
nulls so indexes match `inputRowMeta`. `getLastIncomingFieldValues` drops nulls
(`removeAll`) and uses `Object.toString()`. A null in an earlier column shifts
every later field, dates and booleans change format (`Date.toString()` /
`Boolean.toString()` instead of the value meta's conversion mask, often
`Y`/`N`), and binary becomes `[B@...`. `passParametersToPipeline` then looks
values up by field index; a short list throws `IndexOutOfBoundsException`,
which that method catches and turns into an empty parameter. Group-time and
filen
ame-switch executions are worse: attempt 0 is given the current row, which is
not in the buffer yet, and the retry is given the previous buffered row.
Explicit field mappings therefore change between attempts. The only fallback to
the previous list is an empty or all-null last row.
**Suggestion:** Keep the attempt-0 list for every retry unless a new input
row has actually been accepted. If a re-read is required, build it with
`getInputRowMeta().getString(row, i)` for every column, including nulls, and
use the same row `executePipeline` was called with. Do not route retries
through `getLastIncomingFieldValues()`.
##########
plugins/transforms/pipelineexecutor/src/main/java/org/apache/hop/pipeline/transforms/pipelineexecutor/PipelineExecutor.java:
##########
@@ -330,6 +339,87 @@ Result executeWithRetries(List<String> parameterValues)
throws HopException {
return result;
}
+ @VisibleForTesting
+ void refreshRetriedParameterVariablesFromExternalScopes() {
+ if (!meta.isInheritingAllVariables()) {
+ return;
+ }
+
+ List<PipelineExecutorParameters> parameters = meta.getParameters();
+ if (parameters == null) {
+ return;
+ }
+
+ for (PipelineExecutorParameters parameter : parameters) {
+ if (parameter == null) {
+ continue;
+ }
+
+ // Explicit field/static mappings remain authoritative.
+ if (!Utils.isEmpty(Const.trim(parameter.getField()))
+ || !Utils.isEmpty(Const.trim(parameter.getInput()))) {
+ continue;
+ }
+
+ String variableName = Const.trim(parameter.getVariable());
+ if (Utils.isEmpty(variableName)) {
+ continue;
+ }
+
+ String latestValue =
resolveRetriedVariableValueFromExternalScopes(variableName);
+ if (latestValue != null) {
+ setVariable(variableName, latestValue);
+ }
+ }
+ }
+
+ private String resolveRetriedVariableValueFromExternalScopes(String
variableName) {
+ IPipelineEngine<PipelineMeta> pipeline = getPipeline();
+ if (pipeline != null) {
+ String pipelineValue = pipeline.getVariable(variableName);
+ if (pipelineValue != null) {
+ return pipelineValue;
Review Comment:
**[bug]** `resolveRetriedVariableValueFromExternalScopes` returns the first
non-null `getPipeline().getVariable(name)`. `Variables.getVariable` reads only
that engine's local map, and `Pipeline.initializeFrom` /
`BaseTransform.initializeFrom` copy parent variables and JVM properties into
that map at startup. A later update that Hop does not also write onto this
pipeline — `ActionSetVariables` on the parent workflow, or `System.setProperty`
— stays invisible, because the parent walk (line 385) and the
`System.getProperty` fallback (line 393) run only when the pipeline map has no
such key. After a normal start the key is present, so those branches are dead
for the inherited variable this method is meant to refresh. `getParentEngine`
also stops at the first non-engine variable space, so it never consults
`Pipeline.getParentVariables()` / `Workflow.getParentVariables()`, which fall
back to `variables.getParentVariables()` for a top-level engine. The JVM unit
test passes only becau
se `new LocalPipelineEngine()` never calls `initializeFrom`, leaving an empty
map so the fallback is reached. In a running pipeline that same test would keep
the stale copied value.
**Suggestion:** Do not treat the current pipeline's copied value as proof
that nothing newer exists. Walk `getParentVariables()` (not only parent
engines) and re-read `System.getProperty` when deciding which scope was updated
after the transform's snapshot. Add a test that `initializeFrom`s the pipeline
with the old token, updates the parent workflow and the system property, and
asserts the transform variable becomes the new token.
##########
plugins/transforms/pipelineexecutor/src/test/java/org/apache/hop/pipeline/transforms/pipelineexecutor/PipelineExecutorTest.java:
##########
@@ -478,6 +541,39 @@ void executeWithRetriesRetriesFailedAttemptUntilSuccess()
throws HopException {
verify(executor, times(2)).executePipelineAttempt(any(), anyLong());
}
+ @Test
+ void executeWithRetriesUsesLatestBufferedValuesOnRetryAttempt() throws
HopException {
+ PipelineExecutorMeta meta = new PipelineExecutorMeta();
+ meta.setDefault();
+ meta.setRetryAttempts("1");
+ meta.setRetryDelay("0");
+
+ PipelineExecutor executor = spy(newExecutor(meta, new
PipelineExecutorData()));
+
+ Result failed = new Result();
+ failed.setResult(false);
+ failed.setNrErrors(1);
+ Result success = new Result();
+ success.setResult(true);
+ success.setNrErrors(0);
+
+ doReturn(failed, success).when(executor).executePipelineAttempt(any(),
anyLong());
+ doReturn(Collections.singletonList("latest-value"))
Review Comment:
**[suggestion]** `executeWithRetriesUsesLatestBufferedValuesOnRetryAttempt`
stubs `resolveParameterValuesForRetryAttempt`, and the executor's group buffer
is null, so the test never calls `getLastIncomingFieldValues` or
`passParametersToPipeline`. It only checks that attempt 0 receives the original
list and attempt 1 receives the stub's list. A retry that shifts nulls, changes
a date, or swaps in the wrong row still passes.
`refreshRetriedParameterVariablesFromExternalScopesUsesLatestJvmValueForUnmappedParameter`
(line 327) sets the system property on a pipeline that was not
`initializeFrom`'d, which is the one shape for which the JVM fallback works,
and neither refresh test asserts a parent workflow, an initialized pipeline
snapshot, or that `executeWithRetries` actually calls the refresh before the
next attempt.
**Suggestion:** Drop the stub and drive `executeWithRetries` with a one-row
buffer that contains a null and a date (or a boolean). Assert the list passed
to attempt 1 is still positional `getString` values. For variables,
`initializeFrom` a parent workflow and the pipeline with the old token, change
the parent after attempt 0, and assert the child parameter on attempt 1.
##########
plugins/transforms/pipelineexecutor/src/main/java/org/apache/hop/pipeline/transforms/pipelineexecutor/PipelineExecutor.java:
##########
@@ -330,6 +339,87 @@ Result executeWithRetries(List<String> parameterValues)
throws HopException {
return result;
}
+ @VisibleForTesting
+ void refreshRetriedParameterVariablesFromExternalScopes() {
+ if (!meta.isInheritingAllVariables()) {
+ return;
+ }
+
+ List<PipelineExecutorParameters> parameters = meta.getParameters();
Review Comment:
**[bug]** The refresh only `setVariable`s parameters-grid rows that have an
empty field and an empty static input. That is not how inherited values reach
the child. `createInternalPipeline` passes this transform to
`PipelineEngineFactory.createPipelineEngine`, which `copyFrom`s the whole
transform variable space and `initializeFrom`s the child before
`passParametersToPipeline` runs. `SubExecutionParameters.activate` then, when
inherit-all is on (the meta default), copies every child parameter that is not
listed on the grid from `this.getVariable(name)`. Those names are never
refreshed, so a retry still starts from the transform snapshot taken in
`prepareExecution`. Skipping rows that have a static input also leaves a static
cell of `'${ACCESS_TOKEN}'` resolving against that same stale transform value
(`activate` resolves mapping values with `parent.resolve`). The comment at line
358 does not hold for field mappings either, because issue 1 replaces the
incoming field list on retry.
**Suggestion:** Before creating the child engine, re-sync the transform
variable space from the live parent scopes (the same names `initializeFrom`
copied), then let `createInternalPipeline` and `SubExecutionParameters` pass
that space through. Keep explicit non-variable field and literal static values
as the values captured for attempt 0. Cover an inherit-all child parameter that
is not on the grid, and a grid variable that is only referenced as `'${NAME}'`
in a static cell.
##########
plugins/transforms/pipelineexecutor/src/main/java/org/apache/hop/pipeline/transforms/pipelineexecutor/PipelineExecutor.java:
##########
@@ -330,6 +339,87 @@ Result executeWithRetries(List<String> parameterValues)
throws HopException {
return result;
}
+ @VisibleForTesting
+ void refreshRetriedParameterVariablesFromExternalScopes() {
+ if (!meta.isInheritingAllVariables()) {
+ return;
+ }
+
+ List<PipelineExecutorParameters> parameters = meta.getParameters();
+ if (parameters == null) {
+ return;
+ }
+
+ for (PipelineExecutorParameters parameter : parameters) {
+ if (parameter == null) {
+ continue;
+ }
+
+ // Explicit field/static mappings remain authoritative.
+ if (!Utils.isEmpty(Const.trim(parameter.getField()))
+ || !Utils.isEmpty(Const.trim(parameter.getInput()))) {
+ continue;
+ }
+
+ String variableName = Const.trim(parameter.getVariable());
+ if (Utils.isEmpty(variableName)) {
+ continue;
+ }
+
+ String latestValue =
resolveRetriedVariableValueFromExternalScopes(variableName);
+ if (latestValue != null) {
+ setVariable(variableName, latestValue);
+ }
+ }
Review Comment:
**[nit]** The closing brace of
`refreshRetriedParameterVariablesFromExternalScopes` has a long run of trailing
spaces. `spotless:check` uses google-java-format, which rejects that line, so
CI fails before the retry tests run.
**Suggestion:** Delete the trailing whitespace on that line (or run `./mvnw
spotless:apply` on this module).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
