Skip to content

Commit b377234

Browse files
manan164manan164
authored
fix(core): reset nested JOIN tasks transitively after sub-workflow restart (#1212)
1 parent fb42287 commit b377234

20 files changed

Lines changed: 365 additions & 225 deletions

‎.github/workflows/ci.yml‎

Lines changed: 90 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,31 @@ on:
3737
default: false
3838

3939
jobs:
40+
detect-changes:
41+
runs-on: ubuntu-latest
42+
outputs:
43+
heavy-persistence: ${{ steps.filter.outputs.heavy-persistence }}
44+
steps:
45+
- uses: actions/checkout@v7
46+
with:
47+
ref: ${{ github.event.pull_request.head.sha }}
48+
fetch-depth: 0
49+
- uses: dorny/paths-filter@v3
50+
id: filter
51+
with:
52+
filters: |
53+
heavy-persistence:
54+
- 'cassandra-persistence/**'
55+
- 'es6-persistence/**'
56+
- 'es7-persistence/**'
57+
- 'es8-persistence/**'
58+
- 'mysql-persistence/**'
59+
- 'os-persistence/**'
60+
- 'os-persistence-v2/**'
61+
- 'os-persistence-v3/**'
62+
- 'scheduler/cassandra-persistence/**'
63+
- 'scheduler/mysql-persistence/**'
64+
4065
build:
4166
runs-on: ubuntu-latest
4267
steps:
@@ -70,44 +95,18 @@ jobs:
7095
~/.gradle/wrapper
7196
key: ${{ runner.os }}-gradle-${{ hashFiles('**/*.gradle*', '**/gradle-wrapper.properties') }}
7297
restore-keys: ${{ runner.os }}-gradle-
73-
- name: Force Docker API Version
74-
run: echo 'api.version=1.44' > ~/.docker-java.properties
75-
- name: Cache Docker images
76-
uses: actions/cache@v5
77-
id: docker-cache
78-
with:
79-
path: /tmp/docker-images-build.tar
80-
key: docker-build-v2
81-
- name: Load cached Docker images
82-
if: steps.docker-cache.outputs.cache-hit == 'true'
83-
run: docker load -i /tmp/docker-images-build.tar || true
8498
- name: Build with Gradle
8599
if: github.ref != 'refs/heads/main'
86100
env:
87101
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
88102
SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
89103
run: |
90-
./gradlew build -x :conductor-test-harness:test --scan
104+
./gradlew build -x :conductor-test-harness:test -x test
91105
- name: Build and Publish snapshot
92106
if: github.event_name != 'pull_request' && github.ref == 'refs/heads/main'
93107
run: |
94108
echo "Running build for commit ${{ github.sha }}"
95-
./gradlew build -x :conductor-test-harness:test
96-
- name: Save Docker images for cache
97-
if: steps.docker-cache.outputs.cache-hit != 'true'
98-
run: |
99-
set -euo pipefail
100-
docker image prune -f
101-
mapfile -t images < <((docker images --format '{{.Repository}}:{{.Tag}}' \
102-
| grep -v '<none>' \
103-
| grep -E '(^|/)(elasticsearch|redis|postgres|mysql|mongo|cassandra)(:|/)|mockserver/mockserver|opensearchproject/opensearch|testcontainers/|orkesio/') || true)
104-
if [ "${#images[@]}" -eq 0 ]; then
105-
echo "No Testcontainers-related images to cache; writing empty tar for cache action."
106-
tar -cf /tmp/docker-images-build.tar --files-from /dev/null
107-
exit 0
108-
fi
109-
printf '%s\n' "${images[@]}"
110-
docker save -o /tmp/docker-images-build.tar "${images[@]}"
109+
./gradlew build -x :conductor-test-harness:test -x test
111110
- name: Generate aggregated coverage report
112111
if: always()
113112
run: ./gradlew jacocoAggregatedReport -x test || true
@@ -127,11 +126,71 @@ jobs:
127126
with:
128127
name: coverage-report
129128
path: build/reports/jacoco/aggregated
130-
- name: Store Buildscan URL
129+
unit-test:
130+
needs: detect-changes
131+
runs-on: ubuntu-latest
132+
steps:
133+
- uses: actions/checkout@v7
134+
with:
135+
ref: ${{ github.event.pull_request.head.sha }}
136+
fetch-depth: 0
137+
- name: Free disk space
138+
run: |
139+
sudo rm -rf /usr/share/dotnet
140+
sudo rm -rf /usr/local/lib/android
141+
sudo rm -rf /opt/ghc
142+
- name: Set up Zulu JDK 21
143+
uses: actions/setup-java@v5
144+
with:
145+
distribution: "zulu"
146+
java-version: "21"
147+
- name: Cache Gradle packages
148+
uses: actions/cache@v5
149+
with:
150+
path: |
151+
~/.gradle/caches
152+
~/.gradle/wrapper
153+
key: ${{ runner.os }}-gradle-${{ hashFiles('**/*.gradle*', '**/gradle-wrapper.properties') }}
154+
restore-keys: ${{ runner.os }}-gradle-
155+
- name: Force Docker API Version
156+
run: echo 'api.version=1.44' > ~/.docker-java.properties
157+
- name: Run unit tests
158+
run: |
159+
# On main (post-merge): run everything
160+
if [ "${{ github.event_name }}" != "pull_request" ]; then
161+
./gradlew test -x :conductor-test-harness:test
162+
exit 0
163+
fi
164+
# On PRs: skip heavy container tests (cassandra, es, mysql, opensearch)
165+
# unless their code changed. Redis, postgres, sqlite always run.
166+
if [ "${{ needs.detect-changes.outputs.heavy-persistence }}" == "true" ]; then
167+
./gradlew test -x :conductor-test-harness:test
168+
else
169+
./gradlew test \
170+
-x :conductor-test-harness:test \
171+
-x :conductor-cassandra-persistence:test \
172+
-x :conductor-scheduler-cassandra-persistence:test \
173+
-x :conductor-es6-persistence:test \
174+
-x :conductor-es7-persistence:test \
175+
-x :conductor-es8-persistence:test \
176+
-x :conductor-mysql-persistence:test \
177+
-x :conductor-scheduler-mysql-persistence:test \
178+
-x :conductor-os-persistence:test \
179+
-x :conductor-os-persistence-v2:test \
180+
-x :conductor-os-persistence-v3:test
181+
fi
182+
- name: Publish Test Report
183+
uses: mikepenz/action-junit-report@v6
184+
if: always()
185+
with:
186+
report_paths: "**/build/test-results/test/TEST-*.xml"
187+
check_name: Unit Test Report
188+
- name: Upload unit-test reports
131189
uses: actions/upload-artifact@v7
190+
if: always()
132191
with:
133-
name: build-scan
134-
path: "buildscan.log"
192+
name: unit-test-reports
193+
path: "**/build/reports/tests"
135194
test-harness:
136195
runs-on: ubuntu-latest
137196
steps:

‎awssqs-event-queue/src/test/java/com/netflix/conductor/sqs/eventqueue/DefaultEventQueueProcessorTest.java‎

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -116,7 +116,11 @@ public void initProcessor() {
116116
public void shouldUpdateTaskByReferenceName() throws Exception {
117117
defaultEventQueueProcessor.updateByTaskRefName(
118118
"v_0", "t0", new HashMap<>(), Status.COMPLETED);
119-
Uninterruptibles.sleepUninterruptibly(1_000, TimeUnit.MILLISECONDS);
119+
for (int i = 0;
120+
i < 10 && updatedTasks.stream().noneMatch(t -> "t0".equals(t.getTaskId()));
121+
i++) {
122+
Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
123+
}
120124
assertTrue(updatedTasks.stream().anyMatch(task -> "t0".equals(task.getTaskId())));
121125
}
122126

@@ -130,7 +134,11 @@ public void shouldThrowExceptionForUnknownWorkflow() throws Exception {
130134
@Test
131135
public void shouldUpdateTaskByTaskId() throws Exception {
132136
defaultEventQueueProcessor.updateByTaskId("v_2", "t2", new HashMap<>(), Status.COMPLETED);
133-
Uninterruptibles.sleepUninterruptibly(1_000, TimeUnit.MILLISECONDS);
137+
for (int i = 0;
138+
i < 10 && updatedTasks.stream().noneMatch(t -> "t2".equals(t.getTaskId()));
139+
i++) {
140+
Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
141+
}
134142
assertTrue(updatedTasks.stream().anyMatch(task -> "t2".equals(task.getTaskId())));
135143
}
136144

‎core/src/main/java/com/netflix/conductor/core/execution/WorkflowExecutorOps.java‎

Lines changed: 23 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -1332,9 +1332,6 @@ private void resetUnsuccessfulJoinTasksWithActiveBranches(WorkflowModel workflow
13321332
.containsType(TaskType.TASK_TYPE_FORK_JOIN_DYNAMIC)) {
13331333
return;
13341334
}
1335-
1336-
// A rerun/retry/restart can reactivate one fork branch while the previous JOIN is still
1337-
// terminal. Reopen only those JOIN tasks whose dependencies include an active branch.
13381335
Set<String> activeReferenceTaskNames =
13391336
workflow.getTasks().stream()
13401337
.filter(NON_TERMINAL_TASK)
@@ -1343,16 +1340,22 @@ private void resetUnsuccessfulJoinTasksWithActiveBranches(WorkflowModel workflow
13431340
if (activeReferenceTaskNames.isEmpty()) {
13441341
return;
13451342
}
1346-
1347-
workflow.getTasks().stream()
1348-
.filter(UNSUCCESSFUL_JOIN_TASK)
1349-
.filter(task -> hasActiveJoinDependency(task, activeReferenceTaskNames))
1350-
.peek(
1351-
task -> {
1352-
task.setStatus(TaskModel.Status.IN_PROGRESS);
1353-
addTaskToQueue(task);
1354-
})
1355-
.forEach(executionDAOFacade::updateTask);
1343+
// Iteratively expand active set and reset: a reset JOIN (e.g. inner_join) becomes active,
1344+
// allowing outer JOINs that depend on it to be reset in the same or subsequent pass.
1345+
boolean resetOccurred;
1346+
do {
1347+
resetOccurred = false;
1348+
for (TaskModel task : workflow.getTasks()) {
1349+
if (UNSUCCESSFUL_JOIN_TASK.test(task)
1350+
&& hasActiveJoinDependency(task, activeReferenceTaskNames)) {
1351+
task.setStatus(TaskModel.Status.IN_PROGRESS);
1352+
addTaskToQueue(task);
1353+
executionDAOFacade.updateTask(task);
1354+
activeReferenceTaskNames.add(task.getReferenceTaskName());
1355+
resetOccurred = true;
1356+
}
1357+
}
1358+
} while (resetOccurred);
13561359
}
13571360

13581361
private boolean hasActiveJoinDependency(
@@ -1951,8 +1954,13 @@ private boolean rerunWF(
19511954
notifyWorkflowStatusListener(workflow, WorkflowEventType.RETRIED);
19521955

19531956
// update tasks in datastore to update workflow-tasks relationship for archived
1954-
// workflows
1955-
executionDAOFacade.updateTasks(workflow.getTasks());
1957+
// workflows; exclude rerunFromTask, which is updated individually below — writing its
1958+
// stale FAILED state here would race with the sweeper and can re-terminate the parent
1959+
final String rerunTaskId = rerunFromTask.getTaskId();
1960+
executionDAOFacade.updateTasks(
1961+
workflow.getTasks().stream()
1962+
.filter(t -> !t.getTaskId().equals(rerunTaskId))
1963+
.collect(Collectors.toList()));
19561964
// Remove all tasks after the "rerunFromTask"
19571965
List<TaskModel> filteredTasks = new ArrayList<>();
19581966
for (TaskModel task : workflow.getTasks()) {

‎e2e/src/test/java/io/conductor/e2e/control/SubWorkflowTimeoutRetryTests.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -108,7 +108,7 @@ public void test() {
108108
log.info("Started {} ", workflowInstanceId);
109109
pollAndCompleteTask(workflowInstanceId, "integration_task_1", Map.of());
110110
Workflow workflow = workflowClient.getWorkflow(workflowInstanceId, true);
111-
await().atMost(3, TimeUnit.SECONDS)
111+
await().atMost(10, TimeUnit.SECONDS)
112112
.untilAsserted(
113113
() -> {
114114
Workflow workflow1 =

‎grpc/build.gradle‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -72,6 +72,7 @@ protobuf {
7272
all()*.plugins {
7373
grpc {}
7474
}
75+
all()*.dependsOn(':conductor-common:protogen')
7576
}
7677
}
7778

‎test-harness/src/test/groovy/com/netflix/conductor/test/integration/DoWhileSpec.groovy‎

Lines changed: 5 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,6 @@ import com.netflix.conductor.common.run.Workflow
2121
import com.netflix.conductor.common.utils.TaskUtils
2222
import com.netflix.conductor.core.execution.tasks.Join
2323
import com.netflix.conductor.core.execution.tasks.SubWorkflow
24-
import com.netflix.conductor.dao.QueueDAO
2524
import com.netflix.conductor.test.base.AbstractSpecification
2625

2726
import static com.netflix.conductor.common.metadata.tasks.TaskType.TASK_TYPE_SUB_WORKFLOW
@@ -35,9 +34,6 @@ class DoWhileSpec extends AbstractSpecification {
3534
@Autowired
3635
SubWorkflow subWorkflowTask
3736

38-
@Autowired
39-
QueueDAO queueDAO
40-
4137
def setup() {
4238
workflowTestUtil.registerWorkflows('do_while_integration_test.json',
4339
'do_while_multiple_integration_test.json',
@@ -409,8 +405,11 @@ class DoWhileSpec extends AbstractSpecification {
409405
asyncSystemTaskExecutor.execute(joinTask, joinId)
410406

411407
and: "the sub workflow system task is executed"
412-
List<String> polledSubWorkflowIds = queueDAO.pop(TASK_TYPE_SUB_WORKFLOW, 1, 200)
413-
asyncSystemTaskExecutor.execute(subWorkflowTask, polledSubWorkflowIds[0])
408+
def doWhileSubWfTask = workflowExecutionService.getExecutionStatus(workflowInstanceId, true)
409+
.tasks.find { it.taskType == TASK_TYPE_SUB_WORKFLOW && it.status == Task.Status.SCHEDULED }
410+
if (doWhileSubWfTask) {
411+
asyncSystemTaskExecutor.execute(subWorkflowTask, doWhileSubWfTask.taskId)
412+
}
414413

415414
then: "Verify that the task was polled and acknowledged and workflow is in completed state"
416415
verifyPolledAndAcknowledgedTask(polledAndCompletedTask2)

‎test-harness/src/test/groovy/com/netflix/conductor/test/integration/DynamicForkJoinSpec.groovy‎

Lines changed: 5 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -24,16 +24,12 @@ import com.netflix.conductor.common.run.Workflow
2424
import com.netflix.conductor.core.execution.StartWorkflowInput
2525
import com.netflix.conductor.core.execution.tasks.Join
2626
import com.netflix.conductor.core.execution.tasks.SubWorkflow
27-
import com.netflix.conductor.dao.QueueDAO
2827
import com.netflix.conductor.test.base.AbstractSpecification
2928

3029
import spock.lang.Shared
3130

3231
class DynamicForkJoinSpec extends AbstractSpecification {
3332

34-
@Autowired
35-
QueueDAO queueDAO
36-
3733
@Autowired
3834
Join joinTask
3935

@@ -428,8 +424,11 @@ class DynamicForkJoinSpec extends AbstractSpecification {
428424
workflowTestUtil.verifyPolledAndAcknowledgedTask(pollAndCompleteTask1Try1)
429425

430426
and: "verify that workflow has progressed further ahead and new dynamic tasks have been scheduled"
431-
List<String> polledSubWfIds = queueDAO.pop('SUB_WORKFLOW', 1, 200)
432-
asyncSystemTaskExecutor.execute(subWorkflowTask, polledSubWfIds[0])
427+
def dynSubWfTask = workflowExecutionService.getExecutionStatus(workflowInstanceId, true)
428+
.tasks.find { it.taskType == 'SUB_WORKFLOW' && it.status == Task.Status.SCHEDULED }
429+
if (dynSubWfTask) {
430+
asyncSystemTaskExecutor.execute(subWorkflowTask, dynSubWfTask.taskId)
431+
}
433432
with(workflowExecutionService.getExecutionStatus(workflowInstanceId, true)) {
434433
status == Workflow.WorkflowStatus.RUNNING
435434
tasks.size() == 5

‎test-harness/src/test/groovy/com/netflix/conductor/test/integration/ExternalPayloadStorageSpec.groovy‎

Lines changed: 10 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,6 @@ import com.netflix.conductor.common.metadata.tasks.TaskType
2020
import com.netflix.conductor.common.run.Workflow
2121
import com.netflix.conductor.core.execution.tasks.Join
2222
import com.netflix.conductor.core.execution.tasks.SubWorkflow
23-
import com.netflix.conductor.dao.QueueDAO
2423
import com.netflix.conductor.test.base.AbstractSpecification
2524
import com.netflix.conductor.test.utils.MockExternalPayloadStorage
2625
import com.netflix.conductor.test.utils.UserTask
@@ -67,9 +66,6 @@ class ExternalPayloadStorageSpec extends AbstractSpecification {
6766
@Autowired
6867
MockExternalPayloadStorage mockExternalPayloadStorage
6968

70-
@Autowired
71-
QueueDAO queueDAO
72-
7369
def setup() {
7470
workflowTestUtil.registerWorkflows('simple_workflow_1_integration_test.json',
7571
'conditional_system_task_workflow_integration_test.json',
@@ -445,8 +441,11 @@ class ExternalPayloadStorageSpec extends AbstractSpecification {
445441
verifyPolledAndAcknowledgedLargePayloadTask(pollAndCompleteLargePayloadTask)
446442

447443
when: "the subworkflow is started by issuing a system task call"
448-
List<String> polledTaskIds = queueDAO.pop(TASK_TYPE_SUB_WORKFLOW, 1, 200)
449-
asyncSystemTaskExecutor.execute(subWorkflowTask, polledTaskIds[0])
444+
def inlineSubWfTask = workflowExecutionService.getExecutionStatus(workflowInstanceId, true)
445+
.tasks.find { it.taskType == TASK_TYPE_SUB_WORKFLOW && it.status == Task.Status.SCHEDULED }
446+
if (inlineSubWfTask) {
447+
asyncSystemTaskExecutor.execute(subWorkflowTask, inlineSubWfTask.taskId)
448+
}
450449

451450
and: "verify that the 'integration_task1' is complete and the next task is scheduled"
452451
def workflow = workflowExecutionService.getExecutionStatus(workflowInstanceId, true)
@@ -737,8 +736,11 @@ class ExternalPayloadStorageSpec extends AbstractSpecification {
737736
verifyPolledAndAcknowledgedLargePayloadTask(pollAndCompleteLargePayloadTask)
738737

739738
when: "the sub workflow system task is executed"
740-
List<String> polledTaskIds = queueDAO.pop(TASK_TYPE_SUB_WORKFLOW, 1, 200)
741-
asyncSystemTaskExecutor.execute(subWorkflowTask, polledTaskIds[0])
739+
def dynamicSubWfTask = workflowExecutionService.getExecutionStatus(workflowInstanceId, true)
740+
.tasks.find { it.taskType == TASK_TYPE_SUB_WORKFLOW && it.status == Task.Status.SCHEDULED }
741+
if (dynamicSubWfTask) {
742+
asyncSystemTaskExecutor.execute(subWorkflowTask, dynamicSubWfTask.taskId)
743+
}
742744

743745
then: "verify that workflow has progressed further ahead and new dynamic tasks have been scheduled with externalized payloads"
744746
def workflow = workflowExecutionService.getExecutionStatus(workflowInstanceId, true)

‎test-harness/src/test/groovy/com/netflix/conductor/test/integration/FailureWorkflowSpec.groovy‎

Lines changed: 5 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,6 @@ import org.springframework.beans.factory.annotation.Autowired
1717
import com.netflix.conductor.common.metadata.tasks.Task
1818
import com.netflix.conductor.common.run.Workflow
1919
import com.netflix.conductor.core.execution.tasks.SubWorkflow
20-
import com.netflix.conductor.dao.QueueDAO
2120
import com.netflix.conductor.model.WorkflowModel
2221
import com.netflix.conductor.test.base.AbstractSpecification
2322

@@ -36,9 +35,6 @@ class FailureWorkflowSpec extends AbstractSpecification {
3635
@Autowired
3736
SubWorkflow subWorkflowTask
3837

39-
@Autowired
40-
QueueDAO queueDAO
41-
4238
def setup() {
4339
workflowTestUtil.registerWorkflows(
4440
'failure_workflow_for_terminate_task_workflow.json',
@@ -95,8 +91,11 @@ class FailureWorkflowSpec extends AbstractSpecification {
9591
'', workflowInput, null)
9692

9793
and: "the sub workflow system task is executed"
98-
List<String> polledTaskIds = queueDAO.pop(TASK_TYPE_SUB_WORKFLOW, 1, 200)
99-
asyncSystemTaskExecutor.execute(subWorkflowTask, polledTaskIds[0])
94+
def failureSubWfTask = workflowExecutionService.getExecutionStatus(workflowInstanceId, true)
95+
.tasks.find { it.taskType == TASK_TYPE_SUB_WORKFLOW && it.status == Task.Status.SCHEDULED }
96+
if (failureSubWfTask) {
97+
asyncSystemTaskExecutor.execute(subWorkflowTask, failureSubWfTask.taskId)
98+
}
10099

101100
then: "verify that the sub workflow has failed"
102101
def workflow = workflowExecutionService.getExecutionStatus(workflowInstanceId, true)

0 commit comments

Comments
 (0)