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 d193e9054d Add Retry Mechanism for Failed Pipeline Executor Executions
(#8544)
d193e9054d is described below
commit d193e9054d5955374803385259ffd7482e2e1910
Author: Happy <[email protected]>
AuthorDate: Sat Oct 10 19:33:25 2026 +0800
Add Retry Mechanism for Failed Pipeline Executor Executions (#8544)
* 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
---------
Co-authored-by: happy.he <[email protected]>
---
.../pipeline/transforms/pipeline-executor.adoc | 5 +-
.../pipelineexecutor/PipelineExecutor.java | 119 +++++++++---
.../pipelineexecutor/PipelineExecutorDialog.java | 69 +++++++
.../pipelineexecutor/PipelineExecutorMeta.java | 24 +++
.../messages/messages_en_US.properties | 9 +-
.../pipelineexecutor/PipelineExecutorMetaTest.java | 6 +
.../pipelineexecutor/PipelineExecutorTest.java | 206 +++++++++++++++++++++
.../test/resources/pipeline-executor-transform.xml | 4 +
8 files changed, 415 insertions(+), 27 deletions(-)
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/pipeline-executor.adoc
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/pipeline-executor.adoc
index db8b0904e5..556b1c5cdb 100644
---
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/pipeline-executor.adoc
+++
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/pipeline-executor.adoc
@@ -88,7 +88,9 @@ The filename may contain variables (for example, you can use
the built-in Intern
*NOTE:* It must be considered that, by specifying the pipeline file name using
this option, we can experience a little performance penalty because the
pipeline that will be executed will be initialized at runtime and not during
the initialization phase as usual. On the other side this option gives greater
flexibility in specifying the pipeline filename dynamically and this could be
useful for many use-cases.
|Run configuration|Specify the Pipeline Run Configuration to be used for
execution.
-|Stop after (milliseconds)|Maximum time to wait for the child pipeline to
complete. Empty or 0 means wait indefinitely. When the time is reached the
child pipeline is stopped and the execution-results stream reports a failure
(`ExecutionResult=N`, `ExecutionNrErrors>=1`). The value may contain a variable.
+|Stop after (milliseconds)|Maximum time to wait for each child pipeline
execution attempt. Empty or 0 means wait indefinitely. When the time is reached
the child pipeline is stopped and the execution-results stream reports a
failure (`ExecutionResult=N`, `ExecutionNrErrors>=1`) for that attempt. The
value may contain a variable.
+|Retry attempts on failure|Number of retries after a failed child pipeline
execution attempt. `0` means no retries (default), so the transform executes
the child once. The value can use integer shorthand and variables (for example
`1,000`, `100k`, `1.5m`, `1e3`, `'${PIPELINE_RETRY_ATTEMPTS}'`).
+|Retry delay (milliseconds)|Delay between retry attempts. Empty or `0` means
retry immediately. The delay happens only between attempts and is not part of
the per-attempt stop-after timeout. The value can use integer shorthand and
variables (for example `1,000`, `100k`, `1.5m`, `1e3`,
`'${PIPELINE_RETRY_DELAY_MS}'`).
|===
@@ -252,4 +254,3 @@ Here you can specify where to send the result files from
the pipeline execution.
|Result file name field
|Specify the name of the field for the result files.
|===
-
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 59021f26ab..eb2fa79b5a 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
@@ -232,31 +232,39 @@ public class PipelineExecutor extends
BaseTransform<PipelineExecutorMeta, Pipeli
}
pipelineExecutorData.groupTimeStart = System.currentTimeMillis();
- if (first) {
- discardLogLines(pipelineExecutorData);
+ List<String> parameterValues = incomingFieldValues;
+ if (parameterValues == null) {
+ List<String> lastIncomingFieldValues = getLastIncomingFieldValues();
+ parameterValues = !Utils.isEmpty(lastIncomingFieldValues) ?
lastIncomingFieldValues : null;
}
+ // Execute the pipeline with retries if necessary
+ Result result = executeWithRetries(parameterValues);
+
+ collectPipelineResults(result);
+ collectExecutionResults(result);
+ collectExecutionResultFiles(result);
+
+ pipelineExecutorData.groupBuffer.clear();
+ }
+
+ @VisibleForTesting
+ Result executePipelineAttempt(List<String> parameterValues, long timeoutMs)
throws HopException {
+ PipelineExecutorData pipelineExecutorData = getData();
+
IPipelineEngine<PipelineMeta> executorPipeline = createInternalPipeline();
pipelineExecutorData.setExecutorPipeline(executorPipeline);
- if (incomingFieldValues != null) {
- // Pass parameter values
- passParametersToPipeline(incomingFieldValues);
- } else {
- List<String> lastIncomingFieldValues = getLastIncomingFieldValues();
- // incomingFieldValues == null- There are no more rows - Last Case -
pass previous values if
- // exists
- // If not still pass the null parameter values
- passParametersToPipeline(
- !Utils.isEmpty(lastIncomingFieldValues) ? lastIncomingFieldValues :
incomingFieldValues);
- }
+
+ passParametersToPipeline(parameterValues);
// keep track for drill down in HopGui...
getPipeline().addActiveSubPipeline(getTransformName(), executorPipeline);
- Result result = new Result();
- result.setRows(pipelineExecutorData.groupBuffer);
- executorPipeline.setPreviousResult(result);
+ Result previousResult = new Result();
+ previousResult.setRows(pipelineExecutorData.groupBuffer);
+ executorPipeline.setPreviousResult(previousResult);
+ Result result = new Result();
try {
executorPipeline.prepareExecution();
@@ -264,17 +272,16 @@ public class PipelineExecutor extends
BaseTransform<PipelineExecutorMeta, Pipeli
executorPipeline.startThreads();
// Wait a while until we're done with the pipeline
- long timeoutMs = ExecutionWait.parseTimeoutMs(this,
meta.getWaitTimeout());
boolean finishedInTime = ExecutionWait.waitForPipeline(executorPipeline,
timeoutMs);
result = executorPipeline.getResult();
+ if (result == null) {
+ result = new Result();
+ }
if (!finishedInTime) {
logError(
BaseMessages.getString(
PKG, "PipelineExecutor.Log.WaitTimeoutReached",
Long.toString(timeoutMs)));
- if (result == null) {
- result = new Result();
- }
result.setResult(false);
result.setNrErrors(Math.max(1, result.getNrErrors()));
}
@@ -282,13 +289,77 @@ public class PipelineExecutor extends
BaseTransform<PipelineExecutorMeta, Pipeli
logError("An error occurred executing the pipeline: ", e);
result.setResult(false);
result.setNrErrors(1);
+ stopPreviousAttemptPipeline(pipelineExecutorData);
}
+ return result;
+ }
- collectPipelineResults(result);
- collectExecutionResults(result);
- collectExecutionResultFiles(result);
+ private boolean isFailedResult(Result result) {
+ if (result == null) {
+ return true;
+ }
+ return !result.isResult() || result.getNrErrors() > 0;
+ }
- pipelineExecutorData.groupBuffer.clear();
+ @VisibleForTesting
+ Result executeWithRetries(List<String> parameterValues) throws HopException {
+ int retryAttempts = Math.max(0,
Const.toIntExpanded(resolve(meta.getRetryAttempts()), 0));
+ long retryDelayMs = Math.max(0L,
Const.toLongExpanded(resolve(meta.getRetryDelay()), 0L));
+ long configuredTimeoutMs = ExecutionWait.parseTimeoutMs(this,
meta.getWaitTimeout());
+
+ Result result = null;
+ for (int attempt = 0; attempt <= retryAttempts; attempt++) {
+ if (isStopped()) {
+ if (result == null) {
+ result = new Result();
+ result.setResult(false);
+ result.setNrErrors(1);
+ }
+ break;
+ }
+
+ result = executePipelineAttempt(parameterValues, configuredTimeoutMs);
+ if (!isFailedResult(result) || attempt == retryAttempts) {
+ break;
+ }
+ if (!waitRetryDelay(retryDelayMs)) {
+ break;
+ }
+ cleanupPreviousAttemptPipeline(getData());
+ }
+ return result;
+ }
+
+ private boolean waitRetryDelay(long retryDelayMs) {
+ long remainingDelayMs = retryDelayMs;
+ while (remainingDelayMs > 0) {
+ if (isStopped()) {
+ return false;
+ }
+ long sleepMs = Math.min(remainingDelayMs, 100L);
+ try {
+ Thread.sleep(sleepMs);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ logError(BaseMessages.getString(PKG,
"PipelineExecutor.Log.RetryInterrupted"), e);
+ return false;
+ }
+ remainingDelayMs -= sleepMs;
+ }
+ return !isStopped();
+ }
+
+ @VisibleForTesting
+ void cleanupPreviousAttemptPipeline(PipelineExecutorData
pipelineExecutorData) {
+ stopPreviousAttemptPipeline(pipelineExecutorData);
+ discardLogLines(pipelineExecutorData);
+ }
+
+ private void stopPreviousAttemptPipeline(PipelineExecutorData
pipelineExecutorData) {
+ IPipelineEngine<PipelineMeta> executorPipeline =
pipelineExecutorData.getExecutorPipeline();
+ if (executorPipeline != null && executorPipeline.isRunning()) {
+ executorPipeline.stopAll();
+ }
}
@VisibleForTesting
diff --git
a/plugins/transforms/pipelineexecutor/src/main/java/org/apache/hop/pipeline/transforms/pipelineexecutor/PipelineExecutorDialog.java
b/plugins/transforms/pipelineexecutor/src/main/java/org/apache/hop/pipeline/transforms/pipelineexecutor/PipelineExecutorDialog.java
index 45ec3481b7..7811f33b26 100644
---
a/plugins/transforms/pipelineexecutor/src/main/java/org/apache/hop/pipeline/transforms/pipelineexecutor/PipelineExecutorDialog.java
+++
b/plugins/transforms/pipelineexecutor/src/main/java/org/apache/hop/pipeline/transforms/pipelineexecutor/PipelineExecutorDialog.java
@@ -95,6 +95,8 @@ public class PipelineExecutorDialog extends
BaseTransformDialog {
private Label wlWaitTimeout;
private TextVar wWaitTimeout;
+ private TextVar wRetryAttempts;
+ private TextVar wRetryDelay;
private Button wbPipelineNameInField;
@@ -287,6 +289,7 @@ public class PipelineExecutorDialog extends
BaseTransformDialog {
fdWaitTimeout.top = new FormAttachment(wlWaitTimeout, 0, SWT.CENTER);
fdWaitTimeout.right = new FormAttachment(100, 0);
wWaitTimeout.setLayoutData(fdWaitTimeout);
+ wWaitTimeout.addModifyListener(lsMod);
//
// Add a tab folder for the parameters and various input and output
@@ -308,6 +311,7 @@ public class PipelineExecutorDialog extends
BaseTransformDialog {
addParametersTab();
addExecutionResultTab();
addRowGroupTab();
+ addRetryTab(lsMod);
addResultRowsTab();
addResultFilesTab();
@@ -438,6 +442,8 @@ public class PipelineExecutorDialog extends
BaseTransformDialog {
}
wWaitTimeout.setText(Const.NVL(pipelineExecutorMeta.getWaitTimeout(), ""));
+ wRetryAttempts.setText(Const.NVL(pipelineExecutorMeta.getRetryAttempts(),
""));
+ wRetryDelay.setText(Const.NVL(pipelineExecutorMeta.getRetryDelay(), ""));
// throw in a separate thread.
//
@@ -841,6 +847,67 @@ public class PipelineExecutorDialog extends
BaseTransformDialog {
wTabFolder.setSelection(wTab);
}
+ private void addRetryTab(ModifyListener lsMod) {
+ final CTabItem wTab = new CTabItem(wTabFolder, SWT.NONE);
+ wTab.setFont(GuiResource.getInstance().getFontDefault());
+ wTab.setText(BaseMessages.getString(PKG,
"PipelineExecutorDialog.Retry.Title"));
+ wTab.setToolTipText(BaseMessages.getString(PKG,
"PipelineExecutorDialog.Retry.Tooltip"));
+
+ Composite wInputComposite = new Composite(wTabFolder, SWT.NONE);
+ PropsUi.setLook(wInputComposite);
+
+ FormLayout tabLayout = new FormLayout();
+ tabLayout.marginWidth = 15;
+ tabLayout.marginHeight = 15;
+ wInputComposite.setLayout(tabLayout);
+
+ Label wlRetryAttempts = new Label(wInputComposite, SWT.RIGHT);
+ PropsUi.setLook(wlRetryAttempts);
+ wlRetryAttempts.setText(
+ BaseMessages.getString(PKG,
"PipelineExecutorDialog.RetryAttempts.Label"));
+ FormData fdlRetryAttempts = new FormData();
+ fdlRetryAttempts.left = new FormAttachment(0, 0);
+ fdlRetryAttempts.top = new FormAttachment(0, 0);
+ fdlRetryAttempts.right = new FormAttachment(middle, -margin);
+ wlRetryAttempts.setLayoutData(fdlRetryAttempts);
+
+ wRetryAttempts = new TextVar(variables, wInputComposite, SWT.SINGLE |
SWT.LEFT | SWT.BORDER);
+ wRetryAttempts.enableExpandedInteger();
+ PropsUi.setLook(wRetryAttempts);
+ wRetryAttempts.setToolTipText(
+ BaseMessages.getString(PKG,
"PipelineExecutorDialog.RetryAttempts.Tooltip"));
+ FormData fdRetryAttempts = new FormData();
+ fdRetryAttempts.left = new FormAttachment(middle, 0);
+ fdRetryAttempts.top = new FormAttachment(wlRetryAttempts, 0, SWT.CENTER);
+ fdRetryAttempts.right = new FormAttachment(100, 0);
+ wRetryAttempts.setLayoutData(fdRetryAttempts);
+ wRetryAttempts.addModifyListener(lsMod);
+
+ Label wlRetryDelay = new Label(wInputComposite, SWT.RIGHT);
+ PropsUi.setLook(wlRetryDelay);
+ wlRetryDelay.setText(BaseMessages.getString(PKG,
"PipelineExecutorDialog.RetryDelay.Label"));
+ FormData fdlRetryDelay = new FormData();
+ fdlRetryDelay.left = new FormAttachment(0, 0);
+ fdlRetryDelay.top = new FormAttachment(wRetryAttempts, margin);
+ fdlRetryDelay.right = new FormAttachment(middle, -margin);
+ wlRetryDelay.setLayoutData(fdlRetryDelay);
+
+ wRetryDelay = new TextVar(variables, wInputComposite, SWT.SINGLE |
SWT.LEFT | SWT.BORDER);
+ wRetryDelay.enableExpandedInteger();
+ PropsUi.setLook(wRetryDelay);
+ wRetryDelay.setToolTipText(
+ BaseMessages.getString(PKG,
"PipelineExecutorDialog.RetryDelay.Tooltip"));
+ FormData fdRetryDelay = new FormData();
+ fdRetryDelay.left = new FormAttachment(middle, 0);
+ fdRetryDelay.top = new FormAttachment(wlRetryDelay, 0, SWT.CENTER);
+ fdRetryDelay.right = new FormAttachment(100, 0);
+ wRetryDelay.setLayoutData(fdRetryDelay);
+ wRetryDelay.addModifyListener(lsMod);
+
+ wTab.setControl(wInputComposite);
+ wTabFolder.setSelection(wTab);
+ }
+
private void addExecutionResultTab() {
final CTabItem wTab = new CTabItem(wTabFolder, SWT.NONE);
@@ -1235,6 +1302,8 @@ public class PipelineExecutorDialog extends
BaseTransformDialog {
pipelineExecutorMeta.setFilenameField(wPipelineNameField.getText());
pipelineExecutorMeta.setRunConfigurationName(wRunConfiguration.getText());
pipelineExecutorMeta.setWaitTimeout(wWaitTimeout.getText());
+ pipelineExecutorMeta.setRetryAttempts(wRetryAttempts.getText());
+ pipelineExecutorMeta.setRetryDelay(wRetryDelay.getText());
// Load the information on the tabs, optionally do some
// verifications...
diff --git
a/plugins/transforms/pipelineexecutor/src/main/java/org/apache/hop/pipeline/transforms/pipelineexecutor/PipelineExecutorMeta.java
b/plugins/transforms/pipelineexecutor/src/main/java/org/apache/hop/pipeline/transforms/pipelineexecutor/PipelineExecutorMeta.java
index 492783ab80..14653badf7 100644
---
a/plugins/transforms/pipelineexecutor/src/main/java/org/apache/hop/pipeline/transforms/pipelineexecutor/PipelineExecutorMeta.java
+++
b/plugins/transforms/pipelineexecutor/src/main/java/org/apache/hop/pipeline/transforms/pipelineexecutor/PipelineExecutorMeta.java
@@ -87,6 +87,14 @@ public class PipelineExecutorMeta
@HopMetadataProperty(key = "wait_timeout")
private String waitTimeout;
+ /** Number of retries when child pipeline execution fails. */
+ @HopMetadataProperty(key = "retry_attempts", groupKey = "retry")
+ private String retryAttempts;
+
+ /** Delay between retries in milliseconds. */
+ @HopMetadataProperty(key = "retry_delay", groupKey = "retry")
+ private String retryDelay;
+
/** Flag that indicate that pipeline name is specified in a stream's field */
@HopMetadataProperty(key = "filenameInField")
private boolean filenameInField;
@@ -257,6 +265,20 @@ public class PipelineExecutorMeta
if (value != null) {
setInheritingAllVariables("Y".equalsIgnoreCase(value));
}
+
+ // Load retry options from nested <retry> group.
+ Node retryNode = XmlHandler.getSubNode(node, "retry");
+ if (retryNode == null) {
+ return;
+ }
+ String retryAttemptsValue = XmlHandler.getTagValue(retryNode,
"retry_attempts");
+ if (retryAttemptsValue != null) {
+ setRetryAttempts(retryAttemptsValue);
+ }
+ String retryDelayValue = XmlHandler.getTagValue(retryNode, "retry_delay");
+ if (retryDelayValue != null) {
+ setRetryDelay(retryDelayValue);
+ }
}
@Override
@@ -287,6 +309,8 @@ public class PipelineExecutorMeta
executionLogTextField = "ExecutionLogText";
executionLogChannelIdField = "ExecutionLogChannelId";
resultFilesFileNameField = "FileName";
+ retryAttempts = "0";
+ retryDelay = "";
}
void prepareExecutionResultsFields(IRowMeta row, TransformMeta nextTransform)
diff --git
a/plugins/transforms/pipelineexecutor/src/main/resources/org/apache/hop/pipeline/transforms/pipelineexecutor/messages/messages_en_US.properties
b/plugins/transforms/pipelineexecutor/src/main/resources/org/apache/hop/pipeline/transforms/pipelineexecutor/messages/messages_en_US.properties
index 9ddb5bbc55..a8eb747dbd 100644
---
a/plugins/transforms/pipelineexecutor/src/main/resources/org/apache/hop/pipeline/transforms/pipelineexecutor/messages/messages_en_US.properties
+++
b/plugins/transforms/pipelineexecutor/src/main/resources/org/apache/hop/pipeline/transforms/pipelineexecutor/messages/messages_en_US.properties
@@ -19,6 +19,7 @@ PipelineExecutor.Description=This transform executes a Hop
pipeline, sets parame
PipelineExecutor.Exception.GroupFieldNotFound=Group field ''{0}'' could not be
found in the input stream
PipelineExecutor.Name=Pipeline executor
PipelineExecutor.Log.WaitTimeoutReached=Maximum wait time of {0} ms reached,
stopping the pipeline
+PipelineExecutor.Log.RetryInterrupted=Interrupted while waiting before
retrying failed child pipeline execution
PipelineExecutor.UnexpectedError=There was an unexpected error\:
PipelineExecutorDialog.Browse.Label=Browse...
PipelineExecutorDialog.ColumnInfo.Field=Field name
@@ -70,10 +71,16 @@ PipelineExecutorDialog.ResultRows.Title=Result rows
PipelineExecutorDialog.ResultRows.Tooltip=You can specify the target and
specification of the result rows that need to be passed from the pipeline.
PipelineExecutorDialog.RowGroup.Title=Row grouping
PipelineExecutorDialog.RowGroup.Tooltip=Determine how many rows of data are
passed to the pipeline before execution
+PipelineExecutorDialog.Retry.Title=Retry
+PipelineExecutorDialog.Retry.Tooltip=Configure retries for failed child
pipeline executions.
PipelineExecutorDialog.RunConfiguration.Label=Run configuration
PipelineExecutorDialog.RunConfiguration.Tooltip=The pipeline run configuration
used to execute the specified pipeline. Use the buttons to create, edit or open
the selected configuration.
PipelineExecutorDialog.WaitTimeout.Label=Stop after (milliseconds)
-PipelineExecutorDialog.WaitTimeout.Tooltip=Maximum time to wait for the
pipeline to complete, in milliseconds. Empty or 0 means wait indefinitely. When
the time is reached the pipeline is stopped and the execution result is
reported as failed. You can use a variable such as '${MAX_WAIT_MS}'.
+PipelineExecutorDialog.WaitTimeout.Tooltip=Maximum time to wait for each child
pipeline execution attempt, in milliseconds. Empty or 0 means wait
indefinitely. When the time is reached the child pipeline is stopped and that
attempt is reported as failed. You can use a variable such as '${MAX_WAIT_MS}'.
+PipelineExecutorDialog.RetryAttempts.Label=Retry attempts on failure
+PipelineExecutorDialog.RetryAttempts.Tooltip=Number of retries for failed
child pipeline executions. 0 means no retries. You can use a variable such as
'${PIPELINE_RETRY_ATTEMPTS}'.
+PipelineExecutorDialog.RetryDelay.Label=Retry delay (milliseconds)
+PipelineExecutorDialog.RetryDelay.Tooltip=Delay between retry attempts in
milliseconds. Empty or 0 retries immediately. You can use a variable such as
'${PIPELINE_RETRY_DELAY_MS}'.
PipelineExecutorDialog.Shell.Title=Pipeline executor
PipelineExecutorDialog.TransformName.Label=Transform name
PipelineExecutorMeta.CheckResult.NoInputReceived=No input is received from
previous transforms
diff --git
a/plugins/transforms/pipelineexecutor/src/test/java/org/apache/hop/pipeline/transforms/pipelineexecutor/PipelineExecutorMetaTest.java
b/plugins/transforms/pipelineexecutor/src/test/java/org/apache/hop/pipeline/transforms/pipelineexecutor/PipelineExecutorMetaTest.java
index 2f2423506a..5384592300 100644
---
a/plugins/transforms/pipelineexecutor/src/test/java/org/apache/hop/pipeline/transforms/pipelineexecutor/PipelineExecutorMetaTest.java
+++
b/plugins/transforms/pipelineexecutor/src/test/java/org/apache/hop/pipeline/transforms/pipelineexecutor/PipelineExecutorMetaTest.java
@@ -39,6 +39,8 @@ class PipelineExecutorMetaTest {
assertEquals("${PROJECT_HOME}/loops/child-loops-log-counter.hpl",
meta.getFilename());
assertEquals("1500", meta.getWaitTimeout());
+ assertEquals("3", meta.getRetryAttempts());
+ assertEquals("250", meta.getRetryDelay());
assertEquals("execution results",
meta.getExecutionResultTargetTransform());
assertEquals("ExecutionTime", meta.getExecutionTimeField());
assertEquals("ExecutionResult", meta.getExecutionResultField());
@@ -91,6 +93,8 @@ class PipelineExecutorMetaTest {
assertEquals(meta.getResultFilesFileNameField(),
clone.getResultFilesFileNameField());
assertEquals(meta.getExecutorsOutputTransform(),
clone.getExecutorsOutputTransform());
assertEquals(meta.getWaitTimeout(), clone.getWaitTimeout());
+ assertEquals(meta.getRetryAttempts(), clone.getRetryAttempts());
+ assertEquals(meta.getRetryDelay(), clone.getRetryDelay());
assertEquals(meta.getParameters().size(), clone.getParameters().size());
assertEquals(meta.getResultRows().size(), clone.getResultRows().size());
}
@@ -104,6 +108,8 @@ class PipelineExecutorMetaTest {
assertEquals("1", meta.getGroupSize());
assertEquals("", meta.getGroupField());
assertEquals("", meta.getGroupTime());
+ assertEquals("0", meta.getRetryAttempts());
+ assertEquals("", meta.getRetryDelay());
assertEquals("ExecutionTime", meta.getExecutionTimeField());
assertEquals("ExecutionResult", meta.getExecutionResultField());
assertEquals("ExecutionNrErrors", meta.getExecutionNrErrorsField());
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 6dd60614a4..388faf62c6 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
@@ -23,11 +23,17 @@ import static
org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
+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.doNothing;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -123,6 +129,23 @@ class PipelineExecutorTest {
executor.discardLogLines(data);
}
+ @Test
+ void cleanupPreviousAttemptPipelineStopsRunningChildPipeline() {
+ PipelineExecutorMeta meta = new PipelineExecutorMeta();
+ meta.setDefault();
+ PipelineExecutorData data = new PipelineExecutorData();
+ @SuppressWarnings("unchecked")
+ IPipelineEngine<PipelineMeta> previousChild = mock(IPipelineEngine.class);
+ when(previousChild.isRunning()).thenReturn(true);
+ when(previousChild.getLogChannelId()).thenReturn("previous-child-channel");
+ data.setExecutorPipeline(previousChild);
+ PipelineExecutor executor = newExecutor(meta, data);
+
+ executor.cleanupPreviousAttemptPipeline(data);
+
+ verify(previousChild).stopAll();
+ }
+
@Test
void passParametersToPipelineMapsIncomingFieldToChildParameter() throws
HopException {
PipelineExecutorMeta meta = new PipelineExecutorMeta();
@@ -430,4 +453,187 @@ class PipelineExecutorTest {
assertNotNull(loaded);
assertFalse(loaded.getTransforms().isEmpty());
}
+
+ @Test
+ void executeWithRetriesRetriesFailedAttemptUntilSuccess() throws
HopException {
+ PipelineExecutorMeta meta = new PipelineExecutorMeta();
+ meta.setDefault();
+ meta.setRetryAttempts("2");
+ 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());
+
+ Result result = executor.executeWithRetries(Collections.emptyList());
+
+ assertTrue(result.isResult());
+ verify(executor, times(2)).executePipelineAttempt(any(), anyLong());
+ }
+
+ @Test
+ void executeWithRetriesCleansUpPreviousAttemptOnlyWhenRetrying() 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());
+ doNothing().when(executor).cleanupPreviousAttemptPipeline(any());
+
+ Result result = executor.executeWithRetries(Collections.emptyList());
+
+ assertTrue(result.isResult());
+ verify(executor, times(1)).cleanupPreviousAttemptPipeline(any());
+ }
+
+ @Test
+ void executeWithRetriesDoesNotCleanUpWhenNoRetryWillRun() throws
HopException {
+ PipelineExecutorMeta meta = new PipelineExecutorMeta();
+ meta.setDefault();
+ meta.setRetryAttempts("0");
+ meta.setRetryDelay("0");
+
+ PipelineExecutor executor = spy(newExecutor(meta, new
PipelineExecutorData()));
+
+ Result failed = new Result();
+ failed.setResult(false);
+ failed.setNrErrors(1);
+
+ doReturn(failed).when(executor).executePipelineAttempt(any(), anyLong());
+ doNothing().when(executor).cleanupPreviousAttemptPipeline(any());
+
+ Result result = executor.executeWithRetries(Collections.emptyList());
+
+ assertFalse(result.isResult());
+ verify(executor, never()).cleanupPreviousAttemptPipeline(any());
+ }
+
+ @Test
+ void executePipelineAttemptStopsCurrentChildWhenStartThreadsThrows() throws
HopException {
+ PipelineExecutorMeta meta = new PipelineExecutorMeta();
+ meta.setDefault();
+
+ PipelineExecutorData data = new PipelineExecutorData();
+ data.groupBuffer = new ArrayList<>();
+ data.groupBuffer.add(new RowMetaAndData(new RowMeta(), new Object[0]));
+
+ @SuppressWarnings("unchecked")
+ IPipelineEngine<PipelineMeta> child = mock(IPipelineEngine.class);
+ when(child.getLogChannelId()).thenReturn("failing-child-channel");
+ when(child.isRunning()).thenReturn(true);
+ doThrow(new HopException("start failed")).when(child).startThreads();
+
+ PipelineExecutor executor = spy(newExecutor(meta, data));
+ doReturn(child).when(executor).createInternalPipeline();
+ doNothing().when(executor).passParametersToPipeline(any());
+
+ Result result = executor.executePipelineAttempt(Collections.emptyList(),
0);
+
+ assertFalse(result.isResult());
+ verify(child).stopAll();
+ verify(executor, never()).discardLogLines(any());
+ }
+
+ @Test
+ void executeWithRetriesParsesExpandedRetryAttempts() throws HopException {
+ PipelineExecutorMeta meta = new PipelineExecutorMeta();
+ meta.setDefault();
+ meta.setRetryAttempts("1e0");
+ meta.setRetryDelay("0");
+
+ PipelineExecutor executor = spy(newExecutor(meta, new
PipelineExecutorData()));
+
+ Result failed = new Result();
+ failed.setResult(false);
+ failed.setNrErrors(1);
+
+ doReturn(failed).when(executor).executePipelineAttempt(any(), anyLong());
+
+ Result result = executor.executeWithRetries(Collections.emptyList());
+
+ assertFalse(result.isResult());
+ verify(executor, times(2)).executePipelineAttempt(any(), anyLong());
+ }
+
+ @Test
+ void executeWithRetriesDoesNotRetryWhenRetriesAreDisabled() throws
HopException {
+ PipelineExecutorMeta meta = new PipelineExecutorMeta();
+ meta.setDefault();
+ meta.setRetryAttempts("0");
+ meta.setRetryDelay("0");
+
+ PipelineExecutor executor = spy(newExecutor(meta, new
PipelineExecutorData()));
+
+ Result failed = new Result();
+ failed.setResult(false);
+ failed.setNrErrors(1);
+
+ doReturn(failed).when(executor).executePipelineAttempt(any(), anyLong());
+
+ Result result = executor.executeWithRetries(Collections.emptyList());
+
+ assertFalse(result.isResult());
+ verify(executor, times(1)).executePipelineAttempt(any(), anyLong());
+ }
+
+ @Test
+ void executeWithRetriesUsesConfiguredStopAfterForEachAttempt() throws
HopException {
+ PipelineExecutorMeta meta = new PipelineExecutorMeta();
+ meta.setDefault();
+ meta.setWaitTimeout("1");
+ meta.setRetryAttempts("5");
+ meta.setRetryDelay("0");
+
+ PipelineExecutor executor = spy(newExecutor(meta, new
PipelineExecutorData()));
+
+ Result failed = new Result();
+ failed.setResult(false);
+ failed.setNrErrors(1);
+
+ doReturn(failed).when(executor).executePipelineAttempt(any(), anyLong());
+
+ Result result = executor.executeWithRetries(Collections.emptyList());
+
+ assertFalse(result.isResult());
+ verify(executor, times(6)).executePipelineAttempt(any(), eq(1L));
+ }
+
+ @Test
+ void executeWithRetriesDoesNotStartAnotherAttemptAfterTransformStop() throws
HopException {
+ PipelineExecutorMeta meta = new PipelineExecutorMeta();
+ meta.setDefault();
+ meta.setRetryAttempts("5");
+ meta.setRetryDelay("0");
+
+ PipelineExecutor executor = spy(newExecutor(meta, new
PipelineExecutorData()));
+
+ Result failed = new Result();
+ failed.setResult(false);
+ failed.setNrErrors(1);
+
+ doReturn(failed).when(executor).executePipelineAttempt(any(), anyLong());
+ doReturn(false, true).when(executor).isStopped();
+
+ Result result = executor.executeWithRetries(Collections.emptyList());
+
+ assertFalse(result.isResult());
+ verify(executor, times(1)).executePipelineAttempt(any(), anyLong());
+ }
}
diff --git
a/plugins/transforms/pipelineexecutor/src/test/resources/pipeline-executor-transform.xml
b/plugins/transforms/pipelineexecutor/src/test/resources/pipeline-executor-transform.xml
index 1c5423bb56..844a5c3b10 100644
---
a/plugins/transforms/pipelineexecutor/src/test/resources/pipeline-executor-transform.xml
+++
b/plugins/transforms/pipelineexecutor/src/test/resources/pipeline-executor-transform.xml
@@ -29,6 +29,10 @@
</partitioning>
<run_configuration>local</run_configuration>
<wait_timeout>1500</wait_timeout>
+ <retry>
+ <retry_attempts>3</retry_attempts>
+ <retry_delay>250</retry_delay>
+ </retry>
<filename>${PROJECT_HOME}/loops/child-loops-log-counter.hpl</filename>
<filenameInField>N</filenameInField>
<filenameField/>
