mattcasters commented on code in PR #8544:
URL: https://github.com/apache/hop/pull/8544#discussion_r4236909163
##########
plugins/transforms/pipelineexecutor/src/main/java/org/apache/hop/pipeline/transforms/pipelineexecutor/PipelineExecutor.java:
##########
@@ -232,63 +232,130 @@ private void executePipeline(List<String>
incomingFieldValues) throws HopExcepti
}
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();
+ cleanupPreviousAttemptPipeline(pipelineExecutorData);
+
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();
// run pipeline
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()));
}
} catch (HopException e) {
logError("An error occurred executing the pipeline: ", e);
result.setResult(false);
result.setNrErrors(1);
+ cleanupPreviousAttemptPipeline(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;
+ }
+ }
+ 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) {
+ IPipelineEngine<PipelineMeta> executorPipeline =
pipelineExecutorData.getExecutorPipeline();
+ if (executorPipeline != null && executorPipeline.isRunning()) {
+ executorPipeline.stopAll();
+ }
+ discardLogLines(pipelineExecutorData);
Review Comment:
**[bug]** `discardLogLines` runs for every caller of
`cleanupPreviousAttemptPipeline`, and both callers drop a log channel the
execution-results row still needs.
`executePipelineAttempt` calls this at the start of every attempt. Group
size still defaults to 1 and retries default to 0, so the second input row
deletes the previous child's channel. `collectExecutionResults` has already
copied `ExecutionLogText` onto the output row, and the `ExecutionLogChannelId`
on that row then points at a channel `HopLogStore.discardLines` has removed.
Previously this discard ran only when `first` was still true, and `processRow`
clears `first` before `executePipeline`, so those channels stayed available
across groups.
The `HopException` catch calls this again, before `executePipeline` reaches
`collectExecutionResults`. When `prepareExecution` or `startThreads` throws on
the last attempt, the execution-log field is read from a buffer this method
just cleared. That catch used to leave the child log in place.
`executePipelineAttemptStopsCurrentChildWhenStartThreadsThrows` only verifies
`stopAll()`, so it still passes if the log is kept.
**Suggestion:** In the catch, stop the engine when it is still running and
leave its log channel alone. Call `discardLogLines` only when a later retry is
about to start, so the last attempt's log is still there for
`collectExecutionResults`. Leave the previous group's channel in place when the
next input group starts its first attempt.
--
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]
