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 32c2ce6400 Fix:stale parameter handling in the Pipeline Executor when
retries are enabled and variables are inherited from parent scopes (#9059)
32c2ce6400 is described below
commit 32c2ce6400844ce99cfb060b0c7e9cd24884463e
Author: Happy <[email protected]>
AuthorDate: Sun Oct 25 03:56:45 2026 +0800
Fix:stale parameter handling in the Pipeline Executor when retries are
enabled and variables are inherited from parent scopes (#9059)
* Enhance DatabaseJoin to support named and positional SQL parameters
* Enhance DatabaseJoin to support runtime output metadata initialization
and improve parameter handling
* Add tests for mixed named and positional SQL parameters in
DatabaseJoinMeta
* Added a new core SQL parser: parseSqlParameterSpec(String) that returns
* Refactor DatabaseJoin to simplify output field name handling and ensure
parameter specification is initialized
* Refactor DatabaseJoinMeta to streamline SQL parameter handling and
enhance metadata discovery
* Add tests for SQL parameter parsing and metadata handling in
DatabaseJoinMeta
* Fix SQL parameter parsing for JSONB operators and ARRAY expressions
- Skip only explicit JSONB operator tokens: `?|`, `?&`, `??`, `??|`,
and `??&`.
- Consume complete operator tokens so `col ?? ?` still counts the trailing
`?` as a bind parameter.
- Stop treating all `[...]` expressions as bracket-quoted identifiers.
- Add dialect-aware bracket handling for SQL Server.
- Preserve PostgreSQL parameter markers in `ARRAY[?]` and `ARRAY[?, ?]`.
- Apply the same parsing behavior to the Database Join path.
- Expand regression coverage for JSONB operators, `LIKE`, `NOT LIKE`,
`BETWEEN`, `SELECT ? FROM dual`, and ARRAY expressions.
* Enhance parseSqlParameterSpec() call in DatabaseJoin to support
bracket-quoted identifiers
* Refactor parseSqlParameterSpec() in DatabaseJoinMeta to support
bracket-quoted identifiers
* Add tests for mixed named and positional placeholders in SQL parameter
parsing
* Refactor SQL parameter tests to improve clarity and coverage for JSONB
operators and bracket-quoted identifiers
* Add resolved parameters panel to DatabaseJoinDialog for improved SQL
parameter handling
* Refactor PipelineExecutor to improve parameter handling and add retry
logic for pipeline execution
* Add retry attempts and delay configuration to PipelineExecutorDialog for
improved error handling
* Add unit tests for retry logic in PipelineExecutor to validate execution
attempts
* Add retry configuration panel to PipelineExecutorDialog for enhanced
error handling
* Add support for retry options in PipelineExecutorMeta with XML handling
* Add timeout handling to executePipelineAttempt for retry logic
* Add support for additional parameters in executePipelineAttempt and
enhance retry logic tests
* Add log message for interrupted retries and update tooltip for wait
timeout
* Refactor retry logic in executeWithRetries and add cleanup for previous
attempts
* Add tests for cleanup and retry logic in PipelineExecutor
* Update documentation for pipeline execution parameters to clarify retry
behavior and timing
* Refactor cleanup logic in executePipelineAttempt to use
stopPreviousAttemptPipeline
* Add tests for cleanup behavior in executeWithRetries method
* Enhance retry logic in PipelineExecutor to refresh parameter variables
from external scopes
* Add tests for refreshing parameter variables from external scopes in
PipelineExecutor
* Refactor PipelineExecutor to streamline parameter variable refresh logic
and improve retry handling
* Add tests for refreshing parameter variables from external scopes in
PipelineExecutor
* Add integration tests for Pipeline Executor retries
Run a child that fails once, then succeeds. Check that an inherited
variable and a static ${TOKEN} cell pick up the value published on the
root scope, and that a null, a boolean and a date stay the same on both
attempts.
---------
Co-authored-by: happy.he <[email protected]>
Co-authored-by: mattcasters <[email protected]>
---
.../transforms/0125-retry-fields-child.hpl | 245 +++++++++++++
.../transforms/0125-retry-fields-parent.hpl | 175 +++++++++
.../transforms/0125-retry-inherited-child.hpl | 224 ++++++++++++
.../transforms/0125-retry-inherited-parent.hpl | 112 ++++++
.../transforms/0125-retry-static-child.hpl | 224 ++++++++++++
.../transforms/0125-retry-static-parent.hpl | 117 ++++++
.../files/expected/0125-retry-fields.txt | 2 +
.../files/expected/0125-retry-inherited.txt | 2 +
.../files/expected/0125-retry-static.txt | 2 +
.../main-0125-pipeline-executor-retry.hwf | 394 +++++++++++++++++++++
.../pipelineexecutor/PipelineExecutor.java | 80 ++++-
.../pipelineexecutor/PipelineExecutorTest.java | 267 +++++++++++++-
12 files changed, 1830 insertions(+), 14 deletions(-)
diff --git a/integration-tests/transforms/0125-retry-fields-child.hpl
b/integration-tests/transforms/0125-retry-fields-child.hpl
new file mode 100644
index 0000000000..8694f385a7
--- /dev/null
+++ b/integration-tests/transforms/0125-retry-fields-child.hpl
@@ -0,0 +1,245 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+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.
+
+-->
+<pipeline>
+ <info>
+ <name>0125-retry-fields-child</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description>Records field-mapped parameters. The first attempt fails so
the executor retries the same row.</description>
+ <extended_description/>
+ <pipeline_version/>
+ <pipeline_type>Normal</pipeline_type>
+ <parameters>
+ <parameter>
+ <name>LABEL</name>
+ <default_value/>
+ <description/>
+ </parameter>
+ <parameter>
+ <name>NOTE</name>
+ <default_value/>
+ <description/>
+ </parameter>
+ <parameter>
+ <name>FLAG</name>
+ <default_value/>
+ <description/>
+ </parameter>
+ <parameter>
+ <name>WHEN</name>
+ <default_value/>
+ <description/>
+ </parameter>
+ </parameters>
+ <capture_transform_performance>N</capture_transform_performance>
+
<transform_performance_capturing_delay>1000</transform_performance_capturing_delay>
+
<transform_performance_capturing_size_limit>100</transform_performance_capturing_size_limit>
+ <created_user>-</created_user>
+ <created_date>2026/10/24 00:00:00.000</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2026/10/24 00:00:00.000</modified_date>
+ </info>
+ <notepads>
+ </notepads>
+ <order>
+ <hop>
+ <from>Generate rows</from>
+ <to>Record fields</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>Record fields</from>
+ <to>Needs retry</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>Needs retry</from>
+ <to>Abort</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>Needs retry</from>
+ <to>Done</to>
+ <enabled>Y</enabled>
+ </hop>
+ </order>
+ <transform>
+ <name>Generate rows</name>
+ <type>RowGenerator</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <fields>
+ <field>
+ <name>row</name>
+ <type>Integer</type>
+ <format>0</format>
+ <currency/>
+ <decimal/>
+ <group/>
+ <nullif>1</nullif>
+ <length>-1</length>
+ <precision>0</precision>
+ <set_empty_string>N</set_empty_string>
+ </field>
+ </fields>
+ <interval_in_ms>5000</interval_in_ms>
+ <last_time_field/>
+ <limit>1</limit>
+ <never_ending>N</never_ending>
+ <row_time_field/>
+ <attributes/>
+ <GUI>
+ <xloc>80</xloc>
+ <yloc>96</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Record fields</name>
+ <type>ScriptValueMod</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <jsScripts>
+ <jsScript>
+ <jsScript_type>0</jsScript_type>
+ <jsScript_name>Script 1</jsScript_name>
+ <jsScript_script>var label = getVariable("LABEL", "");
+var note = getVariable("NOTE", "");
+var flag = getVariable("FLAG", "");
+var when = getVariable("WHEN", "");
+var path = java.lang.System.getProperty("java.io.tmpdir") +
"/hop-it-0125-fields.txt";
+var out = new java.io.FileWriter(path, true);
+out.write(label + "|" + note + "|" + flag + "|" + when + "\n");
+out.close();
+var marker = new java.io.File(java.lang.System.getProperty("java.io.tmpdir") +
"/hop-it-0125-fields.marker");
+var first = !marker.exists();
+if (first) {
+ var touch = new java.io.FileWriter(marker);
+ touch.close();
+}
+var fail = first ? "Y" : "N";</jsScript_script>
+ </jsScript>
+ </jsScripts>
+ <fields>
+ <field>
+ <name>fail</name>
+ <rename>fail</rename>
+ <type>String</type>
+ <length>-1</length>
+ <precision>-1</precision>
+ <replace>N</replace>
+ </field>
+ </fields>
+ <optimizationLevel>9</optimizationLevel>
+ <languageVersion>ES6</languageVersion>
+ <attributes/>
+ <GUI>
+ <xloc>256</xloc>
+ <yloc>96</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Needs retry</name>
+ <type>FilterRows</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <send_true_to>Abort</send_true_to>
+ <send_false_to>Done</send_false_to>
+ <compare>
+ <condition>
+ <negated>N</negated>
+ <leftvalue>fail</leftvalue>
+ <function>=</function>
+ <rightvalue/>
+ <value>
+ <name>constant</name>
+ <type>String</type>
+ <text>Y</text>
+ <length>-1</length>
+ <precision>-1</precision>
+ <isnull>N</isnull>
+ <mask/>
+ </value>
+ </condition>
+ </compare>
+ <attributes/>
+ <GUI>
+ <xloc>432</xloc>
+ <yloc>96</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Abort</name>
+ <type>Abort</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <row_threshold>0</row_threshold>
+ <message>field parameters recorded, retry the same row</message>
+ <always_log_rows>Y</always_log_rows>
+ <abort_option>ABORT_WITH_ERROR</abort_option>
+ <attributes/>
+ <GUI>
+ <xloc>608</xloc>
+ <yloc>32</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Done</name>
+ <type>Dummy</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <attributes/>
+ <GUI>
+ <xloc>608</xloc>
+ <yloc>160</yloc>
+ </GUI>
+ </transform>
+ <transform_error_handling>
+ </transform_error_handling>
+ <attributes/>
+</pipeline>
diff --git a/integration-tests/transforms/0125-retry-fields-parent.hpl
b/integration-tests/transforms/0125-retry-fields-parent.hpl
new file mode 100644
index 0000000000..afc7b9ecb8
--- /dev/null
+++ b/integration-tests/transforms/0125-retry-fields-parent.hpl
@@ -0,0 +1,175 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+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.
+
+-->
+<pipeline>
+ <info>
+ <name>0125-retry-fields-parent</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description>Field mappings include a null column, a boolean and a date.
Both attempts must receive the same converted values.</description>
+ <extended_description/>
+ <pipeline_version/>
+ <pipeline_type>Normal</pipeline_type>
+ <parameters>
+ </parameters>
+ <capture_transform_performance>N</capture_transform_performance>
+
<transform_performance_capturing_delay>1000</transform_performance_capturing_delay>
+
<transform_performance_capturing_size_limit>100</transform_performance_capturing_size_limit>
+ <created_user>-</created_user>
+ <created_date>2026/10/24 00:00:00.000</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2026/10/24 00:00:00.000</modified_date>
+ </info>
+ <notepads>
+ </notepads>
+ <order>
+ <hop>
+ <from>Generate rows</from>
+ <to>Execute child</to>
+ <enabled>Y</enabled>
+ </hop>
+ </order>
+ <transform>
+ <name>Generate rows</name>
+ <type>RowGenerator</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <fields>
+ <field>
+ <name>label</name>
+ <type>String</type>
+ <format/>
+ <currency/>
+ <decimal/>
+ <group/>
+ <nullif>alpha</nullif>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ </field>
+ <field>
+ <name>note</name>
+ <type>String</type>
+ <format/>
+ <currency/>
+ <decimal/>
+ <group/>
+ <nullif/>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ </field>
+ <field>
+ <name>flag</name>
+ <type>Boolean</type>
+ <format>Y/N</format>
+ <currency/>
+ <decimal/>
+ <group/>
+ <nullif>Y</nullif>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ </field>
+ <field>
+ <name>when</name>
+ <type>Date</type>
+ <format>yyyy-MM-dd</format>
+ <currency/>
+ <decimal/>
+ <group/>
+ <nullif>2026-10-21</nullif>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ </field>
+ </fields>
+ <interval_in_ms>5000</interval_in_ms>
+ <last_time_field/>
+ <limit>1</limit>
+ <never_ending>N</never_ending>
+ <row_time_field/>
+ <attributes/>
+ <GUI>
+ <xloc>80</xloc>
+ <yloc>96</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Execute child</name>
+ <type>PipelineExecutor</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <run_configuration>local</run_configuration>
+ <filename>${PROJECT_HOME}/0125-retry-fields-child.hpl</filename>
+ <filenameInField>N</filenameInField>
+ <filenameField/>
+ <group_size>1</group_size>
+ <group_field/>
+ <group_time/>
+ <parameters>
+ <variable_mapping>
+ <variable>LABEL</variable>
+ <field>label</field>
+ <input/>
+ </variable_mapping>
+ <variable_mapping>
+ <variable>NOTE</variable>
+ <field>note</field>
+ <input/>
+ </variable_mapping>
+ <variable_mapping>
+ <variable>FLAG</variable>
+ <field>flag</field>
+ <input/>
+ </variable_mapping>
+ <variable_mapping>
+ <variable>WHEN</variable>
+ <field>when</field>
+ <input/>
+ </variable_mapping>
+ </parameters>
+ <inherit_all_vars>N</inherit_all_vars>
+ <retry>
+ <retry_attempts>1</retry_attempts>
+ <retry_delay>0</retry_delay>
+ </retry>
+ <execution_result_target_transform/>
+ <result_files_file_name_field>FileName</result_files_file_name_field>
+ <attributes/>
+ <GUI>
+ <xloc>288</xloc>
+ <yloc>96</yloc>
+ </GUI>
+ </transform>
+ <transform_error_handling>
+ </transform_error_handling>
+ <attributes/>
+</pipeline>
diff --git a/integration-tests/transforms/0125-retry-inherited-child.hpl
b/integration-tests/transforms/0125-retry-inherited-child.hpl
new file mode 100644
index 0000000000..277a5f8ae3
--- /dev/null
+++ b/integration-tests/transforms/0125-retry-inherited-child.hpl
@@ -0,0 +1,224 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+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.
+
+-->
+<pipeline>
+ <info>
+ <name>0125-retry-inherited-child</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description>Records inherited TOKEN, then publishes a new root value and
fails until that value arrives.</description>
+ <extended_description/>
+ <pipeline_version/>
+ <pipeline_type>Normal</pipeline_type>
+ <parameters>
+ <parameter>
+ <name>TOKEN</name>
+ <default_value/>
+ <description>Inherited from the parent Pipeline Executor. Empty until
the parent passes a value.</description>
+ </parameter>
+ </parameters>
+ <capture_transform_performance>N</capture_transform_performance>
+
<transform_performance_capturing_delay>1000</transform_performance_capturing_delay>
+
<transform_performance_capturing_size_limit>100</transform_performance_capturing_size_limit>
+ <created_user>-</created_user>
+ <created_date>2026/10/24 00:00:00.000</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2026/10/24 00:00:00.000</modified_date>
+ </info>
+ <notepads>
+ </notepads>
+ <order>
+ <hop>
+ <from>Generate rows</from>
+ <to>Record token</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>Record token</from>
+ <to>Needs retry</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>Needs retry</from>
+ <to>Abort</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>Needs retry</from>
+ <to>Done</to>
+ <enabled>Y</enabled>
+ </hop>
+ </order>
+ <transform>
+ <name>Generate rows</name>
+ <type>RowGenerator</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <fields>
+ <field>
+ <name>row</name>
+ <type>Integer</type>
+ <format>0</format>
+ <currency/>
+ <decimal/>
+ <group/>
+ <nullif>1</nullif>
+ <length>-1</length>
+ <precision>0</precision>
+ <set_empty_string>N</set_empty_string>
+ </field>
+ </fields>
+ <interval_in_ms>5000</interval_in_ms>
+ <last_time_field/>
+ <limit>1</limit>
+ <never_ending>N</never_ending>
+ <row_time_field/>
+ <attributes/>
+ <GUI>
+ <xloc>80</xloc>
+ <yloc>96</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Record token</name>
+ <type>ScriptValueMod</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <jsScripts>
+ <jsScript>
+ <jsScript_type>0</jsScript_type>
+ <jsScript_name>Script 1</jsScript_name>
+ <jsScript_script>var token = getVariable("TOKEN", "");
+var path = java.lang.System.getProperty("java.io.tmpdir") +
"/hop-it-0125-inherited.txt";
+var out = new java.io.FileWriter(path, true);
+out.write(token + "\n");
+out.close();
+if (token != "fresh") {
+ setVariable("TOKEN", "fresh", "r");
+}
+var fail = token == "fresh" ? "N" : "Y";</jsScript_script>
+ </jsScript>
+ </jsScripts>
+ <fields>
+ <field>
+ <name>fail</name>
+ <rename>fail</rename>
+ <type>String</type>
+ <length>-1</length>
+ <precision>-1</precision>
+ <replace>N</replace>
+ </field>
+ </fields>
+ <optimizationLevel>9</optimizationLevel>
+ <languageVersion>ES6</languageVersion>
+ <attributes/>
+ <GUI>
+ <xloc>256</xloc>
+ <yloc>96</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Needs retry</name>
+ <type>FilterRows</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <send_true_to>Abort</send_true_to>
+ <send_false_to>Done</send_false_to>
+ <compare>
+ <condition>
+ <negated>N</negated>
+ <leftvalue>fail</leftvalue>
+ <function>=</function>
+ <rightvalue/>
+ <value>
+ <name>constant</name>
+ <type>String</type>
+ <text>Y</text>
+ <length>-1</length>
+ <precision>-1</precision>
+ <isnull>N</isnull>
+ <mask/>
+ </value>
+ </condition>
+ </compare>
+ <attributes/>
+ <GUI>
+ <xloc>432</xloc>
+ <yloc>96</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Abort</name>
+ <type>Abort</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <row_threshold>0</row_threshold>
+ <message>inherited token is not fresh yet</message>
+ <always_log_rows>Y</always_log_rows>
+ <abort_option>ABORT_WITH_ERROR</abort_option>
+ <attributes/>
+ <GUI>
+ <xloc>608</xloc>
+ <yloc>32</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Done</name>
+ <type>Dummy</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <attributes/>
+ <GUI>
+ <xloc>608</xloc>
+ <yloc>160</yloc>
+ </GUI>
+ </transform>
+ <transform_error_handling>
+ </transform_error_handling>
+ <attributes/>
+</pipeline>
diff --git a/integration-tests/transforms/0125-retry-inherited-parent.hpl
b/integration-tests/transforms/0125-retry-inherited-parent.hpl
new file mode 100644
index 0000000000..7273c7e470
--- /dev/null
+++ b/integration-tests/transforms/0125-retry-inherited-parent.hpl
@@ -0,0 +1,112 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+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.
+
+-->
+<pipeline>
+ <info>
+ <name>0125-retry-inherited-parent</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description>Pipeline Executor retries a child that is not given TOKEN on
the parameters grid. Inherit-all must pick up the value the child publishes on
the root scope.</description>
+ <extended_description/>
+ <pipeline_version/>
+ <pipeline_type>Normal</pipeline_type>
+ <parameters>
+ <parameter>
+ <name>TOKEN</name>
+ <default_value>stale</default_value>
+ <description>Starting value. The child replaces it on the root scope
during the failed attempt.</description>
+ </parameter>
+ </parameters>
+ <capture_transform_performance>N</capture_transform_performance>
+
<transform_performance_capturing_delay>1000</transform_performance_capturing_delay>
+
<transform_performance_capturing_size_limit>100</transform_performance_capturing_size_limit>
+ <created_user>-</created_user>
+ <created_date>2026/10/24 00:00:00.000</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2026/10/24 00:00:00.000</modified_date>
+ </info>
+ <notepads>
+ </notepads>
+ <order>
+ <hop>
+ <from>Generate rows</from>
+ <to>Execute child</to>
+ <enabled>Y</enabled>
+ </hop>
+ </order>
+ <transform>
+ <name>Generate rows</name>
+ <type>RowGenerator</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <fields>
+ </fields>
+ <interval_in_ms>5000</interval_in_ms>
+ <last_time_field/>
+ <limit>1</limit>
+ <never_ending>N</never_ending>
+ <row_time_field/>
+ <attributes/>
+ <GUI>
+ <xloc>80</xloc>
+ <yloc>96</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Execute child</name>
+ <type>PipelineExecutor</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <run_configuration>local</run_configuration>
+ <filename>${PROJECT_HOME}/0125-retry-inherited-child.hpl</filename>
+ <filenameInField>N</filenameInField>
+ <filenameField/>
+ <group_size>1</group_size>
+ <group_field/>
+ <group_time/>
+ <parameters>
+ </parameters>
+ <inherit_all_vars>Y</inherit_all_vars>
+ <retry>
+ <retry_attempts>1</retry_attempts>
+ <retry_delay>0</retry_delay>
+ </retry>
+ <execution_result_target_transform/>
+ <result_files_file_name_field>FileName</result_files_file_name_field>
+ <attributes/>
+ <GUI>
+ <xloc>288</xloc>
+ <yloc>96</yloc>
+ </GUI>
+ </transform>
+ <transform_error_handling>
+ </transform_error_handling>
+ <attributes/>
+</pipeline>
diff --git a/integration-tests/transforms/0125-retry-static-child.hpl
b/integration-tests/transforms/0125-retry-static-child.hpl
new file mode 100644
index 0000000000..86644cbf81
--- /dev/null
+++ b/integration-tests/transforms/0125-retry-static-child.hpl
@@ -0,0 +1,224 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+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.
+
+-->
+<pipeline>
+ <info>
+ <name>0125-retry-static-child</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description>Records parameter P. P is a static ${TOKEN} cell on the
parent executor.</description>
+ <extended_description/>
+ <pipeline_version/>
+ <pipeline_type>Normal</pipeline_type>
+ <parameters>
+ <parameter>
+ <name>P</name>
+ <default_value/>
+ <description>Value of the static ${TOKEN} cell.</description>
+ </parameter>
+ </parameters>
+ <capture_transform_performance>N</capture_transform_performance>
+
<transform_performance_capturing_delay>1000</transform_performance_capturing_delay>
+
<transform_performance_capturing_size_limit>100</transform_performance_capturing_size_limit>
+ <created_user>-</created_user>
+ <created_date>2026/10/24 00:00:00.000</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2026/10/24 00:00:00.000</modified_date>
+ </info>
+ <notepads>
+ </notepads>
+ <order>
+ <hop>
+ <from>Generate rows</from>
+ <to>Record token</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>Record token</from>
+ <to>Needs retry</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>Needs retry</from>
+ <to>Abort</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>Needs retry</from>
+ <to>Done</to>
+ <enabled>Y</enabled>
+ </hop>
+ </order>
+ <transform>
+ <name>Generate rows</name>
+ <type>RowGenerator</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <fields>
+ <field>
+ <name>row</name>
+ <type>Integer</type>
+ <format>0</format>
+ <currency/>
+ <decimal/>
+ <group/>
+ <nullif>1</nullif>
+ <length>-1</length>
+ <precision>0</precision>
+ <set_empty_string>N</set_empty_string>
+ </field>
+ </fields>
+ <interval_in_ms>5000</interval_in_ms>
+ <last_time_field/>
+ <limit>1</limit>
+ <never_ending>N</never_ending>
+ <row_time_field/>
+ <attributes/>
+ <GUI>
+ <xloc>80</xloc>
+ <yloc>96</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Record token</name>
+ <type>ScriptValueMod</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <jsScripts>
+ <jsScript>
+ <jsScript_type>0</jsScript_type>
+ <jsScript_name>Script 1</jsScript_name>
+ <jsScript_script>var token = getVariable("P", "");
+var path = java.lang.System.getProperty("java.io.tmpdir") +
"/hop-it-0125-static.txt";
+var out = new java.io.FileWriter(path, true);
+out.write(token + "\n");
+out.close();
+if (token != "fresh") {
+ setVariable("TOKEN", "fresh", "r");
+}
+var fail = token == "fresh" ? "N" : "Y";</jsScript_script>
+ </jsScript>
+ </jsScripts>
+ <fields>
+ <field>
+ <name>fail</name>
+ <rename>fail</rename>
+ <type>String</type>
+ <length>-1</length>
+ <precision>-1</precision>
+ <replace>N</replace>
+ </field>
+ </fields>
+ <optimizationLevel>9</optimizationLevel>
+ <languageVersion>ES6</languageVersion>
+ <attributes/>
+ <GUI>
+ <xloc>256</xloc>
+ <yloc>96</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Needs retry</name>
+ <type>FilterRows</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <send_true_to>Abort</send_true_to>
+ <send_false_to>Done</send_false_to>
+ <compare>
+ <condition>
+ <negated>N</negated>
+ <leftvalue>fail</leftvalue>
+ <function>=</function>
+ <rightvalue/>
+ <value>
+ <name>constant</name>
+ <type>String</type>
+ <text>Y</text>
+ <length>-1</length>
+ <precision>-1</precision>
+ <isnull>N</isnull>
+ <mask/>
+ </value>
+ </condition>
+ </compare>
+ <attributes/>
+ <GUI>
+ <xloc>432</xloc>
+ <yloc>96</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Abort</name>
+ <type>Abort</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <row_threshold>0</row_threshold>
+ <message>static token is not fresh yet</message>
+ <always_log_rows>Y</always_log_rows>
+ <abort_option>ABORT_WITH_ERROR</abort_option>
+ <attributes/>
+ <GUI>
+ <xloc>608</xloc>
+ <yloc>32</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Done</name>
+ <type>Dummy</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <attributes/>
+ <GUI>
+ <xloc>608</xloc>
+ <yloc>160</yloc>
+ </GUI>
+ </transform>
+ <transform_error_handling>
+ </transform_error_handling>
+ <attributes/>
+</pipeline>
diff --git a/integration-tests/transforms/0125-retry-static-parent.hpl
b/integration-tests/transforms/0125-retry-static-parent.hpl
new file mode 100644
index 0000000000..c7e2976b83
--- /dev/null
+++ b/integration-tests/transforms/0125-retry-static-parent.hpl
@@ -0,0 +1,117 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+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.
+
+-->
+<pipeline>
+ <info>
+ <name>0125-retry-static-parent</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description>The executor passes P as the static cell ${TOKEN}. A retry
must resolve that cell again after the child updates TOKEN.</description>
+ <extended_description/>
+ <pipeline_version/>
+ <pipeline_type>Normal</pipeline_type>
+ <parameters>
+ <parameter>
+ <name>TOKEN</name>
+ <default_value>stale</default_value>
+ <description>Resolved by the static parameter cell. The child replaces
it on the root scope.</description>
+ </parameter>
+ </parameters>
+ <capture_transform_performance>N</capture_transform_performance>
+
<transform_performance_capturing_delay>1000</transform_performance_capturing_delay>
+
<transform_performance_capturing_size_limit>100</transform_performance_capturing_size_limit>
+ <created_user>-</created_user>
+ <created_date>2026/10/24 00:00:00.000</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2026/10/24 00:00:00.000</modified_date>
+ </info>
+ <notepads>
+ </notepads>
+ <order>
+ <hop>
+ <from>Generate rows</from>
+ <to>Execute child</to>
+ <enabled>Y</enabled>
+ </hop>
+ </order>
+ <transform>
+ <name>Generate rows</name>
+ <type>RowGenerator</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <fields>
+ </fields>
+ <interval_in_ms>5000</interval_in_ms>
+ <last_time_field/>
+ <limit>1</limit>
+ <never_ending>N</never_ending>
+ <row_time_field/>
+ <attributes/>
+ <GUI>
+ <xloc>80</xloc>
+ <yloc>96</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Execute child</name>
+ <type>PipelineExecutor</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <run_configuration>local</run_configuration>
+ <filename>${PROJECT_HOME}/0125-retry-static-child.hpl</filename>
+ <filenameInField>N</filenameInField>
+ <filenameField/>
+ <group_size>1</group_size>
+ <group_field/>
+ <group_time/>
+ <parameters>
+ <variable_mapping>
+ <variable>P</variable>
+ <field/>
+ <input>${TOKEN}</input>
+ </variable_mapping>
+ </parameters>
+ <inherit_all_vars>Y</inherit_all_vars>
+ <retry>
+ <retry_attempts>1</retry_attempts>
+ <retry_delay>0</retry_delay>
+ </retry>
+ <execution_result_target_transform/>
+ <result_files_file_name_field>FileName</result_files_file_name_field>
+ <attributes/>
+ <GUI>
+ <xloc>288</xloc>
+ <yloc>96</yloc>
+ </GUI>
+ </transform>
+ <transform_error_handling>
+ </transform_error_handling>
+ <attributes/>
+</pipeline>
diff --git a/integration-tests/transforms/files/expected/0125-retry-fields.txt
b/integration-tests/transforms/files/expected/0125-retry-fields.txt
new file mode 100644
index 0000000000..a53e7f526f
--- /dev/null
+++ b/integration-tests/transforms/files/expected/0125-retry-fields.txt
@@ -0,0 +1,2 @@
+alpha||Y|2026-10-21
+alpha||Y|2026-10-21
diff --git
a/integration-tests/transforms/files/expected/0125-retry-inherited.txt
b/integration-tests/transforms/files/expected/0125-retry-inherited.txt
new file mode 100644
index 0000000000..1f8665bf5d
--- /dev/null
+++ b/integration-tests/transforms/files/expected/0125-retry-inherited.txt
@@ -0,0 +1,2 @@
+stale
+fresh
diff --git a/integration-tests/transforms/files/expected/0125-retry-static.txt
b/integration-tests/transforms/files/expected/0125-retry-static.txt
new file mode 100644
index 0000000000..1f8665bf5d
--- /dev/null
+++ b/integration-tests/transforms/files/expected/0125-retry-static.txt
@@ -0,0 +1,2 @@
+stale
+fresh
diff --git a/integration-tests/transforms/main-0125-pipeline-executor-retry.hwf
b/integration-tests/transforms/main-0125-pipeline-executor-retry.hwf
new file mode 100644
index 0000000000..542625c6fd
--- /dev/null
+++ b/integration-tests/transforms/main-0125-pipeline-executor-retry.hwf
@@ -0,0 +1,394 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+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.
+
+-->
+<workflow>
+ <name>main-0125-pipeline-executor-retry</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description>Pipeline Executor retries: an inherited variable, a static
${TOKEN} cell, and field values that include a null, a boolean and a
date.</description>
+ <extended_description/>
+ <workflow_version/>
+ <created_user>-</created_user>
+ <created_date>2026/10/24 00:00:00.000</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2026/10/24 00:00:00.000</modified_date>
+ <parameters>
+ </parameters>
+ <actions>
+ <action>
+ <name>Start</name>
+ <description/>
+ <type>SPECIAL</type>
+ <attributes/>
+ <DayOfMonth>1</DayOfMonth>
+ <hour>12</hour>
+ <intervalMinutes>60</intervalMinutes>
+ <intervalSeconds>0</intervalSeconds>
+ <minutes>0</minutes>
+ <repeat>N</repeat>
+ <schedulerType>0</schedulerType>
+ <weekDay>1</weekDay>
+ <parallel>N</parallel>
+ <xloc>64</xloc>
+ <yloc>64</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>delete inherited result</name>
+ <description/>
+ <type>DELETE_FILE</type>
+ <attributes/>
+ <fail_if_file_not_exists>N</fail_if_file_not_exists>
+ <filename>${java.io.tmpdir}/hop-it-0125-inherited.txt</filename>
+ <parallel>N</parallel>
+ <xloc>224</xloc>
+ <yloc>64</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>inherited variable</name>
+ <description/>
+ <type>PIPELINE</type>
+ <attributes/>
+ <filename>${PROJECT_HOME}/0125-retry-inherited-parent.hpl</filename>
+ <params_from_previous>N</params_from_previous>
+ <exec_per_row>N</exec_per_row>
+ <clear_rows>N</clear_rows>
+ <clear_files>N</clear_files>
+ <set_logfile>N</set_logfile>
+ <loglevel>Basic</loglevel>
+ <set_append_logfile>N</set_append_logfile>
+ <wait_until_finished>Y</wait_until_finished>
+ <create_parent_folder>N</create_parent_folder>
+ <run_configuration>local</run_configuration>
+ <parameters>
+ <pass_all_parameters>N</pass_all_parameters>
+ <parameter>
+ <name>TOKEN</name>
+ <stream_name/>
+ <value>stale</value>
+ </parameter>
+ </parameters>
+ <parallel>N</parallel>
+ <xloc>416</xloc>
+ <yloc>64</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>check inherited variable</name>
+ <description/>
+ <type>FILE_COMPARE</type>
+ <attributes/>
+ <filename1>${java.io.tmpdir}/hop-it-0125-inherited.txt</filename1>
+
<filename2>${PROJECT_HOME}/files/expected/0125-retry-inherited.txt</filename2>
+ <add_filename_result>N</add_filename_result>
+ <parallel>N</parallel>
+ <xloc>608</xloc>
+ <yloc>64</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>delete static result</name>
+ <description/>
+ <type>DELETE_FILE</type>
+ <attributes/>
+ <fail_if_file_not_exists>N</fail_if_file_not_exists>
+ <filename>${java.io.tmpdir}/hop-it-0125-static.txt</filename>
+ <parallel>N</parallel>
+ <xloc>800</xloc>
+ <yloc>64</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>static variable cell</name>
+ <description/>
+ <type>PIPELINE</type>
+ <attributes/>
+ <filename>${PROJECT_HOME}/0125-retry-static-parent.hpl</filename>
+ <params_from_previous>N</params_from_previous>
+ <exec_per_row>N</exec_per_row>
+ <clear_rows>N</clear_rows>
+ <clear_files>N</clear_files>
+ <set_logfile>N</set_logfile>
+ <loglevel>Basic</loglevel>
+ <set_append_logfile>N</set_append_logfile>
+ <wait_until_finished>Y</wait_until_finished>
+ <create_parent_folder>N</create_parent_folder>
+ <run_configuration>local</run_configuration>
+ <parameters>
+ <pass_all_parameters>N</pass_all_parameters>
+ <parameter>
+ <name>TOKEN</name>
+ <stream_name/>
+ <value>stale</value>
+ </parameter>
+ </parameters>
+ <parallel>N</parallel>
+ <xloc>992</xloc>
+ <yloc>64</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>check static variable cell</name>
+ <description/>
+ <type>FILE_COMPARE</type>
+ <attributes/>
+ <filename1>${java.io.tmpdir}/hop-it-0125-static.txt</filename1>
+
<filename2>${PROJECT_HOME}/files/expected/0125-retry-static.txt</filename2>
+ <add_filename_result>N</add_filename_result>
+ <parallel>N</parallel>
+ <xloc>1184</xloc>
+ <yloc>64</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>delete field result</name>
+ <description/>
+ <type>DELETE_FILE</type>
+ <attributes/>
+ <fail_if_file_not_exists>N</fail_if_file_not_exists>
+ <filename>${java.io.tmpdir}/hop-it-0125-fields.txt</filename>
+ <parallel>N</parallel>
+ <xloc>1376</xloc>
+ <yloc>64</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>delete field marker</name>
+ <description/>
+ <type>DELETE_FILE</type>
+ <attributes/>
+ <fail_if_file_not_exists>N</fail_if_file_not_exists>
+ <filename>${java.io.tmpdir}/hop-it-0125-fields.marker</filename>
+ <parallel>N</parallel>
+ <xloc>1472</xloc>
+ <yloc>64</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>field values</name>
+ <description/>
+ <type>PIPELINE</type>
+ <attributes/>
+ <filename>${PROJECT_HOME}/0125-retry-fields-parent.hpl</filename>
+ <params_from_previous>N</params_from_previous>
+ <exec_per_row>N</exec_per_row>
+ <clear_rows>N</clear_rows>
+ <clear_files>N</clear_files>
+ <set_logfile>N</set_logfile>
+ <loglevel>Basic</loglevel>
+ <set_append_logfile>N</set_append_logfile>
+ <wait_until_finished>Y</wait_until_finished>
+ <create_parent_folder>N</create_parent_folder>
+ <run_configuration>local</run_configuration>
+ <parameters>
+ <pass_all_parameters>Y</pass_all_parameters>
+ </parameters>
+ <parallel>N</parallel>
+ <xloc>1568</xloc>
+ <yloc>64</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>check field values</name>
+ <description/>
+ <type>FILE_COMPARE</type>
+ <attributes/>
+ <filename1>${java.io.tmpdir}/hop-it-0125-fields.txt</filename1>
+
<filename2>${PROJECT_HOME}/files/expected/0125-retry-fields.txt</filename2>
+ <add_filename_result>N</add_filename_result>
+ <parallel>N</parallel>
+ <xloc>1760</xloc>
+ <yloc>64</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>Success</name>
+ <description/>
+ <type>SUCCESS</type>
+ <attributes/>
+ <parallel>N</parallel>
+ <xloc>1952</xloc>
+ <yloc>64</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>Abort</name>
+ <description/>
+ <type>ABORT</type>
+ <attributes/>
+ <message>Pipeline Executor retry integration test failed</message>
+ <parallel>N</parallel>
+ <xloc>992</xloc>
+ <yloc>240</yloc>
+ <attributes_hac/>
+ </action>
+ </actions>
+ <hops>
+ <hop>
+ <from>Start</from>
+ <to>delete inherited result</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>Y</unconditional>
+ </hop>
+ <hop>
+ <from>delete inherited result</from>
+ <to>inherited variable</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>delete inherited result</from>
+ <to>Abort</to>
+ <enabled>Y</enabled>
+ <evaluation>N</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>inherited variable</from>
+ <to>check inherited variable</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>inherited variable</from>
+ <to>Abort</to>
+ <enabled>Y</enabled>
+ <evaluation>N</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>check inherited variable</from>
+ <to>delete static result</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>check inherited variable</from>
+ <to>Abort</to>
+ <enabled>Y</enabled>
+ <evaluation>N</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>delete static result</from>
+ <to>static variable cell</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>delete static result</from>
+ <to>Abort</to>
+ <enabled>Y</enabled>
+ <evaluation>N</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>static variable cell</from>
+ <to>check static variable cell</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>static variable cell</from>
+ <to>Abort</to>
+ <enabled>Y</enabled>
+ <evaluation>N</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>check static variable cell</from>
+ <to>delete field result</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>check static variable cell</from>
+ <to>Abort</to>
+ <enabled>Y</enabled>
+ <evaluation>N</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>delete field result</from>
+ <to>delete field marker</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>delete field result</from>
+ <to>Abort</to>
+ <enabled>Y</enabled>
+ <evaluation>N</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>delete field marker</from>
+ <to>field values</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>delete field marker</from>
+ <to>Abort</to>
+ <enabled>Y</enabled>
+ <evaluation>N</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>field values</from>
+ <to>check field values</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>field values</from>
+ <to>Abort</to>
+ <enabled>Y</enabled>
+ <evaluation>N</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>check field values</from>
+ <to>Success</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>check field values</from>
+ <to>Abort</to>
+ <enabled>Y</enabled>
+ <evaluation>N</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ </hops>
+ <notepads>
+ </notepads>
+ <attributes/>
+</workflow>
diff --git
a/plugins/transforms/pipelineexecutor/src/main/java/org/apache/hop/pipeline/transforms/pipelineexecutor/PipelineExecutor.java
b/plugins/transforms/pipelineexecutor/src/main/java/org/apache/hop/pipeline/transforms/pipelineexecutor/PipelineExecutor.java
index eb2fa79b5a..7f68413bf7 100644
---
a/plugins/transforms/pipelineexecutor/src/main/java/org/apache/hop/pipeline/transforms/pipelineexecutor/PipelineExecutor.java
+++
b/plugins/transforms/pipelineexecutor/src/main/java/org/apache/hop/pipeline/transforms/pipelineexecutor/PipelineExecutor.java
@@ -20,16 +20,18 @@ package org.apache.hop.pipeline.transforms.pipelineexecutor;
import com.google.common.annotations.VisibleForTesting;
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.Collections;
import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import org.apache.hop.core.Const;
import org.apache.hop.core.IRowSet;
import org.apache.hop.core.Result;
import org.apache.hop.core.ResultFile;
import org.apache.hop.core.RowMetaAndData;
import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.exception.HopValueException;
import org.apache.hop.core.logging.HopLogStore;
import org.apache.hop.core.logging.LoggingRegistry;
import org.apache.hop.core.parameters.SubExecutionParameters;
@@ -37,6 +39,7 @@ import org.apache.hop.core.row.IRowMeta;
import org.apache.hop.core.row.RowDataUtil;
import org.apache.hop.core.row.RowMeta;
import org.apache.hop.core.util.Utils;
+import org.apache.hop.core.variables.IVariables;
import org.apache.hop.execution.ExecutionWait;
import org.apache.hop.i18n.BaseMessages;
import org.apache.hop.pipeline.Pipeline;
@@ -318,6 +321,10 @@ public class PipelineExecutor extends
BaseTransform<PipelineExecutorMeta, Pipeli
break;
}
+ if (attempt > 0) {
+ refreshRetriedParameterVariablesFromExternalScopes();
+ }
+
result = executePipelineAttempt(parameterValues, configuredTimeoutMs);
if (!isFailedResult(result) || attempt == retryAttempts) {
break;
@@ -330,6 +337,49 @@ public class PipelineExecutor extends
BaseTransform<PipelineExecutorMeta, Pipeli
return result;
}
+ @VisibleForTesting
+ void refreshRetriedParameterVariablesFromExternalScopes() {
+ Set<String> variableNamesToRefresh = new
LinkedHashSet<>(Arrays.asList(getVariableNames()));
+ List<PipelineExecutorParameters> parameters = meta.getParameters();
+ if (parameters != null) {
+ for (PipelineExecutorParameters parameter : parameters) {
+ if (parameter == null) {
+ continue;
+ }
+ String parameterVariable = Const.trim(parameter.getVariable());
+ if (!Utils.isEmpty(parameterVariable)) {
+ variableNamesToRefresh.add(parameterVariable);
+ }
+ }
+ }
+
+ for (String variableName : variableNamesToRefresh) {
+ String latestValue =
resolveRetriedVariableValueFromExternalScopes(variableName);
+ if (latestValue != null) {
+ setVariable(variableName, latestValue);
+ }
+ }
+ }
+
+ private String resolveRetriedVariableValueFromExternalScopes(String
variableName) {
+ if (Utils.isEmpty(variableName)) {
+ return null;
+ }
+
+ IPipelineEngine<PipelineMeta> pipeline = getPipeline();
+ IVariables parent = pipeline != null ? pipeline.getParentVariables() :
null;
+ int guard = 0;
+ while (parent != null && guard++ < 100) {
+ String value = parent.getVariable(variableName);
+ if (value != null) {
+ return value;
+ }
+ parent = parent.getParentVariables();
+ }
+
+ return System.getProperty(variableName);
+ }
+
private boolean waitRetryDelay(long retryDelayMs) {
long remainingDelayMs = retryDelayMs;
while (remainingDelayMs > 0) {
@@ -692,21 +742,29 @@ public class PipelineExecutor extends
BaseTransform<PipelineExecutorMeta, Pipeli
return getData().getExecutorPipeline();
}
- protected List<String> getLastIncomingFieldValues() {
+ protected List<String> getLastIncomingFieldValues() throws HopException {
PipelineExecutorData pipelineExecutorData = getData();
- List<String> lastIncomingFieldValues = new ArrayList<>();
- if (pipelineExecutorData == null ||
pipelineExecutorData.groupBuffer.isEmpty()) {
+ if (pipelineExecutorData == null
+ || pipelineExecutorData.groupBuffer == null
+ || pipelineExecutorData.groupBuffer.isEmpty()) {
return null;
}
- int lastIncomingFieldIndex = pipelineExecutorData.groupBuffer.size() - 1;
- ArrayList lastGroupBufferData =
- new ArrayList(
-
Arrays.asList(pipelineExecutorData.groupBuffer.get(lastIncomingFieldIndex).getData()));
- lastGroupBufferData.removeAll(Collections.singleton(null));
+ RowMetaAndData lastBufferedRow =
+
pipelineExecutorData.groupBuffer.get(pipelineExecutorData.groupBuffer.size() -
1);
+ IRowMeta rowMeta = lastBufferedRow.getRowMeta();
+ Object[] rowData = lastBufferedRow.getData();
+ if (rowMeta == null || rowData == null) {
+ return null;
+ }
- for (Object lastGroupBufferDatum : lastGroupBufferData) {
- lastIncomingFieldValues.add(lastGroupBufferDatum.toString());
+ List<String> lastIncomingFieldValues = new ArrayList<>(rowMeta.size());
+ for (int i = 0; i < rowMeta.size(); i++) {
+ try {
+ lastIncomingFieldValues.add(rowMeta.getString(rowData, i));
+ } catch (HopValueException e) {
+ throw new HopException("Error converting buffered row values", e);
+ }
}
return lastIncomingFieldValues;
}
diff --git
a/plugins/transforms/pipelineexecutor/src/test/java/org/apache/hop/pipeline/transforms/pipelineexecutor/PipelineExecutorTest.java
b/plugins/transforms/pipelineexecutor/src/test/java/org/apache/hop/pipeline/transforms/pipelineexecutor/PipelineExecutorTest.java
index 388faf62c6..e0f88c1c14 100644
---
a/plugins/transforms/pipelineexecutor/src/test/java/org/apache/hop/pipeline/transforms/pipelineexecutor/PipelineExecutorTest.java
+++
b/plugins/transforms/pipelineexecutor/src/test/java/org/apache/hop/pipeline/transforms/pipelineexecutor/PipelineExecutorTest.java
@@ -27,9 +27,11 @@ import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
@@ -42,13 +44,17 @@ import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.Date;
import java.util.List;
import java.util.Objects;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.hop.core.HopEnvironment;
import org.apache.hop.core.Result;
import org.apache.hop.core.RowMetaAndData;
import org.apache.hop.core.exception.HopException;
import org.apache.hop.core.row.RowMeta;
+import org.apache.hop.core.row.value.ValueMetaBoolean;
+import org.apache.hop.core.row.value.ValueMetaDate;
import org.apache.hop.core.row.value.ValueMetaString;
import org.apache.hop.core.variables.Variables;
import org.apache.hop.junit.rules.RestoreHopEngineEnvironmentExtension;
@@ -58,9 +64,11 @@ import org.apache.hop.pipeline.TransformWithMappingMeta;
import org.apache.hop.pipeline.engine.IPipelineEngine;
import org.apache.hop.pipeline.engines.local.LocalPipelineEngine;
import org.apache.hop.pipeline.transform.TransformMeta;
+import org.apache.hop.workflow.engines.local.LocalWorkflowEngine;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
+import org.mockito.InOrder;
/** Unit tests for {@link PipelineExecutor} runtime behaviour (parameter
passing, helpers). */
class PipelineExecutorTest {
@@ -88,7 +96,7 @@ class PipelineExecutorTest {
}
@Test
- void getLastIncomingFieldValuesReturnsNullWhenBufferEmpty() {
+ void getLastIncomingFieldValuesReturnsNullWhenBufferEmpty() throws
HopException {
PipelineExecutorMeta meta = new PipelineExecutorMeta();
meta.setDefault();
PipelineExecutorData data = new PipelineExecutorData();
@@ -99,7 +107,7 @@ class PipelineExecutorTest {
}
@Test
- void getLastIncomingFieldValuesReturnsStringValuesFromLastBufferedRow() {
+ void getLastIncomingFieldValuesPreservesNullsFromLastBufferedRow() throws
HopException {
PipelineExecutorMeta meta = new PipelineExecutorMeta();
meta.setDefault();
PipelineExecutorData data = new PipelineExecutorData();
@@ -114,7 +122,29 @@ class PipelineExecutorTest {
List<String> last = executor.getLastIncomingFieldValues();
assertNotNull(last);
- assertEquals(Collections.singletonList("x"), last);
+ assertEquals(Arrays.asList("x", null), last);
+ }
+
+ @Test
+ void getLastIncomingFieldValuesUsesRowMetaStringConversion() throws
HopException {
+ PipelineExecutorMeta meta = new PipelineExecutorMeta();
+ meta.setDefault();
+ PipelineExecutorData data = new PipelineExecutorData();
+ data.groupBuffer = new ArrayList<>();
+ RowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaString("s1"));
+ rowMeta.addValueMeta(new ValueMetaDate("d1"));
+ rowMeta.addValueMeta(new ValueMetaBoolean("b1"));
+ Date date = new Date(1761051000000L);
+ rowMeta.getValueMeta(1).setConversionMask("yyyy-MM-dd");
+ data.groupBuffer.add(new RowMetaAndData(rowMeta, new Object[] {"x", date,
true}));
+
+ PipelineExecutor executor = newExecutor(meta, data);
+ List<String> last = executor.getLastIncomingFieldValues();
+
+ assertNotNull(last);
+ assertEquals(
+ Arrays.asList("x",
rowMeta.getString(data.groupBuffer.get(0).getData(), 1), "Y"), last);
}
@Test
@@ -308,6 +338,140 @@ class PipelineExecutorTest {
assertEquals("B2", child.getVariable("TEST3_PARAMETER2"));
}
+ @Test
+ void
refreshRetriedParameterVariablesFromExternalScopesUsesLatestJvmValueForUnmappedParameter()
{
+ final String variableName = "ACCESS_TOKEN";
+ final String originalProperty = System.getProperty(variableName);
+ try {
+ PipelineExecutorMeta meta = new PipelineExecutorMeta();
+ meta.setDefault();
+ meta.setInheritingAllVariables(true);
+ PipelineExecutorParameters param = new PipelineExecutorParameters();
+ param.setVariable(variableName);
+ param.setField("");
+ param.setInput("");
+ meta.setParameters(new ArrayList<>(Collections.singletonList(param)));
+
+ PipelineExecutor executor = newExecutor(meta, new
PipelineExecutorData());
+ executor.setVariable(variableName, "stale-token");
+ Variables parent = new Variables();
+ parent.initializeFrom(null);
+ parent.setVariable(variableName, "stale-parent-token");
+ executor.getPipeline().initializeFrom(parent);
+ parent.setVariable(variableName, null);
+ System.setProperty(variableName, "mynewtoken");
+
+ executor.refreshRetriedParameterVariablesFromExternalScopes();
+
+ assertEquals("mynewtoken", executor.getVariable(variableName));
+ } finally {
+ if (originalProperty == null) {
+ System.clearProperty(variableName);
+ } else {
+ System.setProperty(variableName, originalProperty);
+ }
+ }
+ }
+
+ @Test
+ void
refreshRetriedParameterVariablesFromExternalScopesPrefersParentPipelineValueOverJvm()
{
+ final String variableName = "ACCESS_TOKEN";
+ final String originalProperty = System.getProperty(variableName);
+ try {
+ PipelineExecutorMeta meta = new PipelineExecutorMeta();
+ meta.setDefault();
+ meta.setInheritingAllVariables(true);
+ PipelineExecutorParameters param = new PipelineExecutorParameters();
+ param.setVariable(variableName);
+ param.setField("");
+ param.setInput("");
+ meta.setParameters(new ArrayList<>(Collections.singletonList(param)));
+
+ PipelineExecutor executor = newExecutor(meta, new
PipelineExecutorData());
+ executor.setVariable(variableName, "stale-token");
+ LocalWorkflowEngine parentWorkflow = new LocalWorkflowEngine();
+ parentWorkflow.initializeFrom(null);
+ parentWorkflow.setVariable(variableName, "parent-token");
+ executor.getPipeline().initializeFrom(parentWorkflow);
+ System.setProperty(variableName, "jvm-token");
+
+ executor.refreshRetriedParameterVariablesFromExternalScopes();
+
+ assertEquals("parent-token", executor.getVariable(variableName));
+ } finally {
+ if (originalProperty == null) {
+ System.clearProperty(variableName);
+ } else {
+ System.setProperty(variableName, originalProperty);
+ }
+ }
+ }
+
+ @Test
+ void
refreshRetriedParameterVariablesFromExternalScopesRefreshesInheritedParameterNotOnGrid()
+ throws HopException {
+ final String variableName = "ACCESS_TOKEN";
+ PipelineExecutorMeta meta = new PipelineExecutorMeta();
+ meta.setDefault();
+ meta.setInheritingAllVariables(true);
+ meta.setParameters(new ArrayList<>());
+
+ PipelineMeta childMeta = new PipelineMeta();
+ childMeta.setName("child");
+ childMeta.addParameterDefinition(variableName, "", "token");
+ LocalPipelineEngine child = new LocalPipelineEngine(childMeta);
+ child.copyParametersFromDefinitions(childMeta);
+
+ PipelineExecutorData data = new PipelineExecutorData();
+ data.setExecutorPipeline(child);
+
+ PipelineExecutor executor = newExecutor(meta, data);
+ executor.setVariable(variableName, "stale-token");
+ executor.getPipeline().setVariable(variableName, "stale-copied-token");
+ Variables parent = new Variables();
+ parent.setVariable(variableName, "fresh-token");
+ executor.getPipeline().setParentVariables(parent);
+
+ executor.refreshRetriedParameterVariablesFromExternalScopes();
+ executor.passParametersToPipeline(Collections.emptyList());
+
+ assertEquals("fresh-token", child.getVariable(variableName));
+ }
+
+ @Test
+ void
refreshRetriedParameterVariablesFromExternalScopesRefreshesVariableUsedInStaticInput()
+ throws HopException {
+ PipelineExecutorMeta meta = new PipelineExecutorMeta();
+ meta.setDefault();
+ meta.setInheritingAllVariables(true);
+ PipelineExecutorParameters param = new PipelineExecutorParameters();
+ param.setVariable("P");
+ param.setField("");
+ param.setInput("${ACCESS_TOKEN}");
+ meta.setParameters(new ArrayList<>(Collections.singletonList(param)));
+
+ PipelineMeta childMeta = new PipelineMeta();
+ childMeta.setName("child");
+ childMeta.addParameterDefinition("P", "", "mapped from static");
+ LocalPipelineEngine child = new LocalPipelineEngine(childMeta);
+ child.copyParametersFromDefinitions(childMeta);
+
+ PipelineExecutorData data = new PipelineExecutorData();
+ data.setExecutorPipeline(child);
+
+ PipelineExecutor executor = newExecutor(meta, data);
+ executor.setVariable("ACCESS_TOKEN", "stale-token");
+ executor.getPipeline().setVariable("ACCESS_TOKEN", "stale-copied-token");
+ Variables parent = new Variables();
+ parent.setVariable("ACCESS_TOKEN", "fresh-token");
+ executor.getPipeline().setParentVariables(parent);
+
+ executor.refreshRetriedParameterVariablesFromExternalScopes();
+ executor.passParametersToPipeline(Collections.emptyList());
+
+ assertEquals("fresh-token", child.getVariable("P"));
+ }
+
@Test
void collectPipelineResultsDoesNotForwardRowsWhenNoResultRowsTarget() throws
HopException {
PipelineExecutorMeta meta = new PipelineExecutorMeta();
@@ -478,6 +642,103 @@ class PipelineExecutorTest {
verify(executor, times(2)).executePipelineAttempt(any(), anyLong());
}
+ @Test
+ void executeWithRetriesKeepsAttemptZeroParameterValuesOnRetryAttempt()
throws HopException {
+ PipelineExecutorMeta meta = new PipelineExecutorMeta();
+ meta.setDefault();
+ meta.setRetryAttempts("1");
+ meta.setRetryDelay("0");
+
+ PipelineExecutorData data = new PipelineExecutorData();
+ data.groupBuffer = new ArrayList<>();
+ RowMeta bufferedRowMeta = new RowMeta();
+ bufferedRowMeta.addValueMeta(new ValueMetaString("col1"));
+ bufferedRowMeta.addValueMeta(new ValueMetaDate("col2"));
+ bufferedRowMeta.addValueMeta(new ValueMetaBoolean("col3"));
+ bufferedRowMeta.getValueMeta(1).setConversionMask("yyyy-MM-dd");
+ data.groupBuffer.add(
+ new RowMetaAndData(bufferedRowMeta, new Object[] {null, new
Date(1761051000000L), true}));
+
+ PipelineExecutor executor = spy(newExecutor(meta, data));
+
+ 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());
+ Result result = executor.executeWithRetries(Arrays.asList("from-row",
null, "N"));
+
+ assertTrue(result.isResult());
+ InOrder attemptOrder = inOrder(executor);
+ attemptOrder
+ .verify(executor)
+ .executePipelineAttempt(eq(Arrays.asList("from-row", null, "N")),
anyLong());
+ attemptOrder
+ .verify(executor)
+ .executePipelineAttempt(eq(Arrays.asList("from-row", null, "N")),
anyLong());
+ }
+
+ @Test
+ void executeWithRetriesRefreshesFromInitializedParentBeforeRetryAttempt()
throws HopException {
+ final String variableName = "ACCESS_TOKEN";
+ PipelineExecutorMeta meta = new PipelineExecutorMeta();
+ meta.setDefault();
+ meta.setInheritingAllVariables(true);
+ meta.setRetryAttempts("1");
+ meta.setRetryDelay("0");
+ PipelineExecutorParameters param = new PipelineExecutorParameters();
+ param.setVariable(variableName);
+ param.setField("");
+ param.setInput("");
+ meta.setParameters(new ArrayList<>(Collections.singletonList(param)));
+
+ PipelineMeta childMeta = new PipelineMeta();
+ childMeta.setName("child");
+ childMeta.addParameterDefinition(variableName, "", "token");
+ LocalPipelineEngine child = new LocalPipelineEngine(childMeta);
+ child.copyParametersFromDefinitions(childMeta);
+
+ PipelineExecutorData data = new PipelineExecutorData();
+ data.setExecutorPipeline(child);
+ PipelineExecutor executor = spy(newExecutor(meta, data));
+
+ LocalWorkflowEngine parentWorkflow = new LocalWorkflowEngine();
+ parentWorkflow.initializeFrom(null);
+ parentWorkflow.setVariable(variableName, "old-token");
+ executor.getPipeline().initializeFrom(parentWorkflow);
+ executor.setVariable(variableName, "old-token");
+
+ List<String> childParameterByAttempt = new ArrayList<>();
+ AtomicInteger attemptCounter = new AtomicInteger();
+ doAnswer(
+ invocation -> {
+ int attempt = attemptCounter.getAndIncrement();
+ executor.passParametersToPipeline(Collections.emptyList());
+ childParameterByAttempt.add(child.getVariable(variableName));
+
+ Result result = new Result();
+ if (attempt == 0) {
+ parentWorkflow.setVariable(variableName, "new-token");
+ result.setResult(false);
+ result.setNrErrors(1);
+ } else {
+ result.setResult(true);
+ result.setNrErrors(0);
+ }
+ return result;
+ })
+ .when(executor)
+ .executePipelineAttempt(any(), anyLong());
+
+ Result result = executor.executeWithRetries(Collections.emptyList());
+
+ assertTrue(result.isResult());
+ assertEquals(Arrays.asList("old-token", "new-token"),
childParameterByAttempt);
+ }
+
@Test
void executeWithRetriesCleansUpPreviousAttemptOnlyWhenRetrying() throws
HopException {
PipelineExecutorMeta meta = new PipelineExecutorMeta();
