beatum commented on code in PR #8544:
URL: https://github.com/apache/hop/pull/8544#discussion_r4237460800
##########
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:
thanks for your time and have a good day!
here is my latest updates:
Changed :
- `executePipelineAttempt()` no longer cleans up at attempt start.
- On `HopException`, it only stops a still-running child
(`stopPreviousAttemptPipeline`) and does not discard logs.
- `executeWithRetries()` now calls `cleanupPreviousAttemptPipeline()` after
a failed attempt, after delay wait succeeds, and only when another retry will
follow.
Tests:
- Added coverage that cleanup runs only when retrying and not when no retry
will run.
- Updated the startThreads-throws test to assert `discardLogLines()` is not
called in that path.
--
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]
