mattcasters commented on code in PR #8544:
URL: https://github.com/apache/hop/pull/8544#discussion_r4229684898
##########
plugins/transforms/pipelineexecutor/src/main/resources/org/apache/hop/pipeline/transforms/pipelineexecutor/messages/messages_en_US.properties:
##########
@@ -70,10 +70,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.RetryAttempts.Label=Retry attempts on failure
Review Comment:
**[suggestion]** The new options are not documented on the Pipeline Executor
page
(`docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/pipeline-executor.adoc`).
That page has an options table and today describes stop-after only.
**Suggestion:** Document retry attempts and retry delay there, including the
default of no retries, and state whether stop-after applies to each attempt or
is a shared budget across attempts and the delay.
##########
plugins/transforms/pipelineexecutor/src/main/java/org/apache/hop/pipeline/transforms/pipelineexecutor/PipelineExecutor.java:
##########
@@ -283,12 +294,55 @@ private void executePipeline(List<String>
incomingFieldValues) throws HopExcepti
result.setResult(false);
result.setNrErrors(1);
}
+ 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.toInt(resolve(meta.getRetryAttempts()), 0));
+ long retryDelayMs = Math.max(0L,
Const.toLong(resolve(meta.getRetryDelay()), 0L));
+ long configuredTimeoutMs = ExecutionWait.parseTimeoutMs(this,
meta.getWaitTimeout());
+ long retryStartTime = System.currentTimeMillis();
+
+ Result result = null;
+ for (int attempt = 0; attempt <= retryAttempts; attempt++) {
+ long timeoutForAttemptMs = configuredTimeoutMs;
+ if (configuredTimeoutMs > 0) {
Review Comment:
**[bug]** Stop-after is folded into one shared budget for every attempt, but
the retry delay does not honor that budget. `Thread.sleep(retryDelayMs)` always
sleeps the full delay, so a stop-after of 1000 ms and a retry delay of 30000 ms
blocks the transform for about 30 s after a fast failure and only then sees
`remainingMs <= 0` and skips the retry. The same budget means a child that was
stopped for hitting the timeout has no time left for another attempt, and a
later attempt can be started with only a few milliseconds and killed
immediately. The dialog tooltip and `pipeline-executor.adoc` still describe
stop-after as the maximum wait for that child pipeline, which is what a single
execution did before this change.
**Suggestion:** Pass the original `parseTimeoutMs` value into every attempt
so stop-after keeps its per-execution meaning. If a total deadline is
intentional, cap the sleep with the remaining time, do not start an attempt
when the remainder is too small to be useful, and update the tooltip and manual
to say the limit covers every attempt and the delay.
##########
plugins/transforms/pipelineexecutor/src/main/java/org/apache/hop/pipeline/transforms/pipelineexecutor/PipelineExecutor.java:
##########
@@ -283,12 +294,55 @@ private void executePipeline(List<String>
incomingFieldValues) throws HopExcepti
result.setResult(false);
result.setNrErrors(1);
}
+ 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.toInt(resolve(meta.getRetryAttempts()), 0));
+ long retryDelayMs = Math.max(0L,
Const.toLong(resolve(meta.getRetryDelay()), 0L));
+ long configuredTimeoutMs = ExecutionWait.parseTimeoutMs(this,
meta.getWaitTimeout());
+ long retryStartTime = System.currentTimeMillis();
+
+ Result result = null;
+ for (int attempt = 0; attempt <= retryAttempts; attempt++) {
+ long timeoutForAttemptMs = configuredTimeoutMs;
+ if (configuredTimeoutMs > 0) {
+ long elapsedMs = System.currentTimeMillis() - retryStartTime;
+ long remainingMs = configuredTimeoutMs - elapsedMs;
+ if (remainingMs <= 0) {
+ if (result == null) {
+ result = new Result();
+ result.setResult(false);
+ result.setNrErrors(1);
+ }
+ break;
+ }
+ timeoutForAttemptMs = remainingMs;
+ }
+
+ result = executePipelineAttempt(parameterValues, timeoutForAttemptMs);
+ if (!isFailedResult(result) || attempt == retryAttempts) {
+ break;
+ }
+ if (retryDelayMs > 0) {
+ try {
+ Thread.sleep(retryDelayMs);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ logError("Interrupted while waiting before retrying failed child
pipeline execution", e);
Review Comment:
**[nit]** This message is hard-coded English. The other user-visible strings
in this transform go through `BaseMessages` and `messages_en_US.properties`.
**Suggestion:** Add a `PipelineExecutor.Log.RetryInterrupted` key and log
that.
##########
plugins/transforms/pipelineexecutor/src/main/java/org/apache/hop/pipeline/transforms/pipelineexecutor/PipelineExecutor.java:
##########
@@ -283,12 +294,55 @@ private void executePipeline(List<String>
incomingFieldValues) throws HopExcepti
result.setResult(false);
result.setNrErrors(1);
}
+ 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.toInt(resolve(meta.getRetryAttempts()), 0));
+ long retryDelayMs = Math.max(0L,
Const.toLong(resolve(meta.getRetryDelay()), 0L));
+ long configuredTimeoutMs = ExecutionWait.parseTimeoutMs(this,
meta.getWaitTimeout());
+ long retryStartTime = System.currentTimeMillis();
+
+ Result result = null;
+ for (int attempt = 0; attempt <= retryAttempts; attempt++) {
+ long timeoutForAttemptMs = configuredTimeoutMs;
+ if (configuredTimeoutMs > 0) {
+ long elapsedMs = System.currentTimeMillis() - retryStartTime;
+ long remainingMs = configuredTimeoutMs - elapsedMs;
+ if (remainingMs <= 0) {
+ if (result == null) {
+ result = new Result();
+ result.setResult(false);
+ result.setNrErrors(1);
+ }
+ break;
+ }
+ timeoutForAttemptMs = remainingMs;
+ }
+
+ result = executePipelineAttempt(parameterValues, timeoutForAttemptMs);
Review Comment:
**[bug]** A parent stop is ignored for the rest of the retry loop.
`Pipeline.stopTransform` sets `stopped` and calls `stopRunning()`, which only
stops `getData().getExecutorPipeline()`. After a failed attempt that engine has
already finished, so the stop does not interrupt `Thread.sleep(retryDelayMs)`
(the transform thread is not interrupted). This call then runs again with no
`isStopped()` check, including when the delay is 0.
**Suggestion:** Return before each attempt when `isStopped()` is true. Wait
for the delay in short slices that also return once the transform is stopped,
and do not launch another child after that.
##########
plugins/transforms/pipelineexecutor/src/main/java/org/apache/hop/pipeline/transforms/pipelineexecutor/PipelineExecutor.java:
##########
@@ -232,49 +232,60 @@ private void executePipeline(List<String>
incomingFieldValues) throws HopExcepti
}
pipelineExecutorData.groupTimeStart = System.currentTimeMillis();
+ 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();
+
if (first) {
Review Comment:
**[suggestion]** Each attempt creates a new engine and registers it as the
active sub-pipeline, but the previous attempt is not stopped or removed from
the logging registry. `discardLogLines` still runs only when `first` is true,
and `processRow` sets `first` to false before `executePipeline`, so this block
does not run between attempts. If `startThreads()` throws, the catch marks the
result failed and the next attempt replaces `executorPipeline` without stopping
the engine that already started. Log buffers from the failed attempts stay
allocated, and the execution-log field only shows the last channel.
**Suggestion:** Before another attempt, stop the previous engine if it is
still running and discard its log channel. Do that from the `HopException`
catch in `executePipelineAttempt` as well, instead of only when `first` is true.
##########
plugins/transforms/pipelineexecutor/src/main/java/org/apache/hop/pipeline/transforms/pipelineexecutor/PipelineExecutor.java:
##########
@@ -283,12 +294,55 @@ private void executePipeline(List<String>
incomingFieldValues) throws HopExcepti
result.setResult(false);
result.setNrErrors(1);
}
+ 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.toInt(resolve(meta.getRetryAttempts()), 0));
Review Comment:
**[bug]** Both dialog fields call `TextVar.enableExpandedInteger()`. That
widget's contract, and the `#` tooltip, is that values such as `1,000`, `100k`,
`1.5m`, and `1e3` are accepted and that variables are resolved before parsing
(`Const.toIntExpanded` / `Const.toLongExpanded`). These lines use `Const.toInt`
and `Const.toLong`, so those forms fail and become the default 0. Retries or
the delay are then silently disabled. Group size in this class already parses
with `Const.toIntExpanded`.
**Suggestion:** Use `Const.toIntExpanded(resolve(meta.getRetryAttempts()),
0)` and `Const.toLongExpanded(resolve(meta.getRetryDelay()), 0L)`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
