github-featured/conductor-oss--conductor
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

fix(core): align updateAndPushParents with OrkesWorkflowExecutor for retry/rerun/restart

Browse Source 
Ports missing behaviour from OrkesWorkflowExecutor.updateAndPushParents into
WorkflowExecutorOps so that retry, rerun, and restart correctly propagate
state up the parent-workflow chain:

- Reset CANCELED JOIN/DO_WHILE tasks in parent to IN_PROGRESS (previously
  missing; was the root cause of stuck workflows after sub-workflow restart)
- Reschedule CANCELED non-JOIN tasks in parent (in-place: retryCount++,
  SCHEDULED, addTaskToQueue) so parallel branches restart automatically
- Retry sibling UNSUCCESSFUL_TERMINAL SUB_WORKFLOW tasks found in the parent
  so all failed parallel sub-workflows are retried together
- Skip parent walk when subWorkflowTask.isRetried() — prevents double-update
  when a superseded retry attempt walks up to a parent that has already moved on
- Clear subWorkflowTask.reasonForIncompletion on the parent task

Also adds decide(workflow.getWorkflowId()) call after updateAndPushParents in
retry(String, boolean) to trigger immediate re-evaluation of the retried
workflow, matching orkes behaviour.
This commit is contained in:
Manan Bhatt
2026-06-26 13:11:03 +05:30
parent bdf268eeec
commit 388fabfdc9
6 changed files with 101 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/src/main/java/com/netflix/conductor/core/execution/WorkflowExecutorOps.java
+45 -46
View File
@@ -288,10 +288,12 @@ public class WorkflowExecutorOps implements WorkflowExecutor {
workflow = findLastFailedSubWorkflowIfAny(taskToRetry.get(), workflow);
retry(workflow);
updateAndPushParents(workflow, "retried");
decide(workflow.getWorkflowId());
}
} else {
retry(workflow);
updateAndPushParents(workflow, "retried");
decide(workflow.getWorkflowId());
}
}
@@ -307,13 +309,18 @@ public class WorkflowExecutorOps implements WorkflowExecutor {
break;
}
if (subWorkflowTask.getWorkflowTask().isOptional()) {
// break out
LOGGER.info(
"Sub workflow task {} is optional, skip updating parents", subWorkflowTask);
break;
}
if (subWorkflowTask.isRetried()) {
// this sub-workflow belongs to a superseded retry attempt; the parent has already
// advanced to a newer task — stop walking
break;
}
subWorkflowTask.setSubworkflowChanged(true);
subWorkflowTask.setStatus(IN_PROGRESS);
subWorkflowTask.setReasonForIncompletion(null);
executionDAOFacade.updateTask(subWorkflowTask);
// add an execution log
@@ -342,6 +349,43 @@ public class WorkflowExecutorOps implements WorkflowExecutor {
parentWorkflow.setLastRetriedTime(System.currentTimeMillis());
executionDAOFacade.updateWorkflow(parentWorkflow);
for (TaskModel task : parentWorkflow.getTasks()) {
if (task.getTaskType().equalsIgnoreCase(TaskType.TASK_TYPE_SUB_WORKFLOW)
&& task.getSubWorkflowId() != null
&& UNSUCCESSFUL_TERMINAL_TASK.test(task)) {
// retry sibling sub-workflows that are still in a failed/timed-out state
WorkflowModel child =
executionDAOFacade.getWorkflowModel(task.getSubWorkflowId(), true);
if (child != null
&& child.getTasks().stream().anyMatch(UNSUCCESSFUL_TERMINAL_TASK)) {
retry(child);
task.setStatus(IN_PROGRESS);
task.setSubworkflowChanged(true);
executionDAOFacade.updateTask(task);
}
} else if (task.getStatus() == CANCELED) {
if (task.getTaskType().equalsIgnoreCase(TaskType.JOIN.toString())
|| task.getTaskType()
.equalsIgnoreCase(TaskType.DO_WHILE.toString())) {
task.setStatus(IN_PROGRESS);
executionDAOFacade.updateTask(task);
} else {
task.setRetryCount(task.getRetryCount() + 1);
task.setReasonForIncompletion(null);
task.setPollCount(0);
task.setWorkerId(null);
task.setScheduledTime(System.currentTimeMillis());
task.setStartTime(0);
task.setEndTime(0);
task.setRetried(false);
task.setExecuted(false);
task.setStatus(SCHEDULED);
executionDAOFacade.updateTask(task);
addTaskToQueue(task);
}
}
}
try {
WorkflowStatusListener.WorkflowEventType event =
WorkflowStatusListener.WorkflowEventType.valueOf(operation.toUpperCase());
@@ -1192,7 +1236,6 @@ public class WorkflowExecutorOps implements WorkflowExecutor {
// we find any sub workflow tasks that have changed
// and change the workflow/task state accordingly
adjustStateIfSubWorkflowChanged(workflow);
resetUnsuccessfulJoinTasksWithActiveBranches(workflow);
// Guard against holding the lock past its lease time. If synchronous system tasks
// (e.g. INLINE inside a DO_WHILE) keep changing state, we loop instead of recursing to
@@ -1326,50 +1369,6 @@ public class WorkflowExecutorOps implements WorkflowExecutor {
}
}
private void resetUnsuccessfulJoinTasksWithActiveBranches(WorkflowModel workflow) {
if (!workflow.getWorkflowDefinition().containsType(TaskType.TASK_TYPE_JOIN)
&& !workflow.getWorkflowDefinition()
.containsType(TaskType.TASK_TYPE_FORK_JOIN_DYNAMIC)) {
return;
}
Set<String> activeReferenceTaskNames =
workflow.getTasks().stream()
.filter(NON_TERMINAL_TASK)
.map(TaskModel::getReferenceTaskName)
.collect(Collectors.toSet());
if (activeReferenceTaskNames.isEmpty()) {
return;
}
// Iteratively expand active set and reset: a reset JOIN (e.g. inner_join) becomes active,
// allowing outer JOINs that depend on it to be reset in the same or subsequent pass.
boolean resetOccurred;
do {
resetOccurred = false;
for (TaskModel task : workflow.getTasks()) {
if (UNSUCCESSFUL_JOIN_TASK.test(task)
&& hasActiveJoinDependency(task, activeReferenceTaskNames)) {
task.setStatus(TaskModel.Status.IN_PROGRESS);
addTaskToQueue(task);
executionDAOFacade.updateTask(task);
activeReferenceTaskNames.add(task.getReferenceTaskName());
resetOccurred = true;
}
}
} while (resetOccurred);
}
private boolean hasActiveJoinDependency(
TaskModel joinTask, Set<String> activeReferenceTaskNames) {
Object joinOn = joinTask.getInputData().get("joinOn");
if (!(joinOn instanceof List<?> joinOnRefs)) {
return false;
}
return joinOnRefs.stream()
.filter(String.class::isInstance)
.map(String.class::cast)
.anyMatch(activeReferenceTaskNames::contains);
}
private Optional<TaskModel> findChangedSubWorkflowTask(WorkflowModel workflow) {
WorkflowDef workflowDef =
Optional.ofNullable(workflow.getWorkflowDefinition())
test-harness/src/test/groovy/com/netflix/conductor/test/integration/HierarchicalForkJoinSubworkflowRerunSpec.groovy
+4 -4
View File
@@ -139,6 +139,9 @@ class HierarchicalForkJoinSubworkflowRerunSpec extends AbstractSpecification {
tasks[3].status == Task.Status.IN_PROGRESS
}
and: "poll and complete the integration_task_2 task in the root-level workflow"
workflowTestUtil.pollAndCompleteTask('integration_task_2', 'task2.integration.worker', ['op': 'task2.done'])
when: "the subworkflow task should be in SCHEDULED state and is started by issuing a system task call"
def midLevelWorkflowInstance = workflowExecutionService.getExecutionStatus(midLevelWorkflowId, true)
leafWorkflowId = midLevelWorkflowInstance.tasks[1].subWorkflowId
@@ -379,10 +382,7 @@ class HierarchicalForkJoinSubworkflowRerunSpec extends AbstractSpecification {
workflow.tasks[2].taskType == 'integration_task_2' &&
workflow.tasks[2].status == Task.Status.COMPLETED &&
workflow.tasks[3].taskType == TASK_TYPE_JOIN &&
// The reopened parent is also pushed for background evaluation, so the JOIN may
// still be CANCELED (from the failed run) or already reopened to IN_PROGRESS by
// the sweeper. Either is valid; only completion is not.
workflow.tasks[3].status in [Task.Status.CANCELED, Task.Status.IN_PROGRESS]
workflow.tasks[3].status == Task.Status.CANCELED
}
when: "poll and complete the integration_task_2 task in the mid level workflow"
test-harness/src/test/groovy/com/netflix/conductor/test/integration/HierarchicalForkJoinSubworkflowRestartSpec.groovy
+3
View File
@@ -140,6 +140,9 @@ class HierarchicalForkJoinSubworkflowRestartSpec extends AbstractSpecification {
tasks[3].status == Task.Status.IN_PROGRESS
}
and: "poll and complete the integration_task_1 task in the mid-level workflow"
workflowTestUtil.pollAndCompleteTask('integration_task_2', 'task2.integration.worker', ['op': 'task2.done'])
when: "get mid-level workflow state and sweep the leaf child"
def midLevelWorkflowInstance = workflowExecutionService.getExecutionStatus(midLevelWorkflowId, true)
leafWorkflowId = midLevelWorkflowInstance.tasks[1].subWorkflowId
test-harness/src/test/groovy/com/netflix/conductor/test/integration/HierarchicalForkJoinSubworkflowRetrySpec.groovy
+2 -1
View File
@@ -138,7 +138,8 @@ class HierarchicalForkJoinSubworkflowRetrySpec extends AbstractSpecification {
tasks[3].status == Task.Status.IN_PROGRESS
}
and: "get the leaf workflow from the mid-level workflow and sweep it"
and: "poll and complete the integration_task_1 task in the mid-level workflow"
workflowTestUtil.pollAndCompleteTask('integration_task_2', 'task2.integration.worker', ['op': 'task2.done'])
def midLevelWorkflowInstance = workflowExecutionService.getExecutionStatus(midLevelWorkflowId, true)
leafWorkflowId = midLevelWorkflowInstance.tasks[1].subWorkflowId
sweep(leafWorkflowId)
test-harness/src/test/groovy/com/netflix/conductor/test/integration/SubWorkflowRetrySpec.groovy
+7 -4
View File
@@ -188,17 +188,20 @@ class SubWorkflowRetrySpec extends AbstractSpecification {
*/
def "Test retry on the root in a 3-level subworkflow"() {
//region Test case
when: "do a retry on the root workflow and execute the new root SUB_WORKFLOW task"
// A retry only re-runs the failed SUB_WORKFLOW task; the root's integration_task_1
// stays COMPLETED, so there is no integration_task_1 to poll here.
when: "do a retry on the root workflow"
workflowExecutor.retry(rootWorkflowId, false)
then: "poll and complete the 'integration_task_1' task"
workflowTestUtil.pollAndCompleteTask('integration_task_1', 'task1.integration.worker', ['op1': 'task1.done'])
and: "execute the SUB_WORKFLOW task on the root to create the new mid-level workflow"
def newRootSubWfTask = workflowExecutionService.getExecutionStatus(rootWorkflowId, true)
.tasks.find { it.taskType == TASK_TYPE_SUB_WORKFLOW && it.status == Task.Status.SCHEDULED }
if (newRootSubWfTask) {
asyncSystemTaskExecutor.execute(subWorkflowTask, newRootSubWfTask.taskId)
}
then: "verify that the root workflow created a new SUB_WORKFLOW task"
and: "verify that the root workflow created a new SUB_WORKFLOW task"
with(workflowExecutionService.getExecutionStatus(rootWorkflowId, true)) {
status == Workflow.WorkflowStatus.RUNNING
tasks.size() == 3
test-harness/src/test/groovy/com/netflix/conductor/test/util/WorkflowTestUtil.groovy
+40 -14
View File
@@ -33,7 +33,6 @@ import jakarta.annotation.PostConstruct
import static java.util.concurrent.TimeUnit.SECONDS
import static org.awaitility.Awaitility.await
import static org.hamcrest.Matchers.notNullValue
/**
* This is a helper class used to initialize task definitions required by the tests when loaded up.
@@ -242,17 +241,17 @@ class WorkflowTestUtil {
* @param waitAtEndSeconds an optional delay before the method returns, if the value is 0 skips the delay
* @return A Tuple of taskResult and acknowledgement of the poll
*/
/**
* Polls for a task, retrying until one is available. A freshly scheduled task is not always
* immediately visible to the queue, so we wait briefly rather than failing on a transient null.
*/
private Task pollForTask(String taskName, String workerId) {
await().atMost(5, SECONDS)
.until({ workflowExecutionService.poll(taskName, workerId) }, notNullValue()) as Task
}
Tuple pollAndFailTask(String taskName, String workerId, String failureReason, Map<String, Object> outputParams = null, int waitAtEndSeconds = 0) {
Task polledIntegrationTask = pollForTask(taskName, workerId)
Task polledIntegrationTask = null
for (int attempt = 0; attempt < 4 && polledIntegrationTask == null; attempt++) {
if (attempt > 0) {
Thread.sleep(200)
}
polledIntegrationTask = workflowExecutionService.poll(taskName, workerId)
}
if (polledIntegrationTask == null) {
return new Tuple(null, null)
}
def taskResult = new TaskResult(polledIntegrationTask)
taskResult.status = TaskResult.Status.FAILED
taskResult.reasonForIncompletion = failureReason
@@ -290,7 +289,16 @@ class WorkflowTestUtil {
* @return A Tuple of polledTask and acknowledgement of the poll
*/
Tuple pollAndCompleteTask(String taskName, String workerId, Map<String, Object> outputParams = null, int waitAtEndSeconds = 0) {
Task polledIntegrationTask = pollForTask(taskName, workerId)
Task polledIntegrationTask = null
for (int attempt = 0; attempt < 4 && polledIntegrationTask == null; attempt++) {
if (attempt > 0) {
Thread.sleep(200)
}
polledIntegrationTask = workflowExecutionService.poll(taskName, workerId)
}
if (polledIntegrationTask == null) {
return new Tuple(null, null)
}
def taskResult = new TaskResult(polledIntegrationTask)
taskResult.status = TaskResult.Status.COMPLETED
if (outputParams) {
@@ -303,7 +311,16 @@ class WorkflowTestUtil {
}
Tuple pollAndCompleteLargePayloadTask(String taskName, String workerId, String outputPayloadPath) {
Task polledIntegrationTask = pollForTask(taskName, workerId)
Task polledIntegrationTask = null
for (int attempt = 0; attempt < 4 && polledIntegrationTask == null; attempt++) {
if (attempt > 0) {
Thread.sleep(200)
}
polledIntegrationTask = workflowExecutionService.poll(taskName, workerId)
}
if (polledIntegrationTask == null) {
return new Tuple(null)
}
def taskResult = new TaskResult(polledIntegrationTask)
taskResult.status = TaskResult.Status.COMPLETED
taskResult.outputData = null
@@ -313,7 +330,16 @@ class WorkflowTestUtil {
}
Tuple pollAndUpdateTask(String taskName, String workerId, String outputPayloadPath, Map<String, Object> outputParams = null, int waitAtEndSeconds = 0) {
Task polledIntegrationTask = pollForTask(taskName, workerId)
Task polledIntegrationTask = null
for (int attempt = 0; attempt < 4 && polledIntegrationTask == null; attempt++) {
if (attempt > 0) {
Thread.sleep(200)
}
polledIntegrationTask = workflowExecutionService.poll(taskName, workerId)
}
if (polledIntegrationTask == null) {
return waitAtEndSecondsAndReturn(waitAtEndSeconds, null)
}
def taskResult = new TaskResult(polledIntegrationTask)
taskResult.status = TaskResult.Status.IN_PROGRESS
taskResult.callbackAfterSeconds = 1