fix: prevent circular dependency and wrong startNodes in backfill dependent workflow

- Problem 1: Replace single-level self-dep check with ThreadLocal visited set
  to detect and skip indirect circular dependencies (A→B→A), preventing
  StackOverflowError when allLevelDependent=true
- Problem 2: Set startNodes=null for downstream workflows; upstream task node
  codes are not valid in a different workflow definition
- Add tests for OFFLINE skip, not-found skip, and startNodes null assertion

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: 2 people

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/executor/workflow/BackfillWorkflowExecutorDelegate.java

@@ -41,7 +41,9 @@
 import java.time.ZonedDateTime;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
 import java.util.stream.Collectors;
 
 import lombok.extern.slf4j.Slf4j;
@@ -55,6 +57,8 @@
 @Component
 public class BackfillWorkflowExecutorDelegate implements IExecutorDelegate<BackfillWorkflowDTO, List<Integer>> {
 
+    private static final ThreadLocal<Set<Long>> BACKFILL_VISITING_WORKFLOWS = new ThreadLocal<>();
+
     @Autowired
     private CommandDao commandDao;
 
@@ -183,89 +187,104 @@ private Integer doBackfillWorkflow(final BackfillWorkflowDTO backfillWorkflowDTO
 
     private void doBackfillDependentWorkflow(final BackfillWorkflowDTO backfillWorkflowDTO,
                                              final List<String> backfillTimeList) {
-        // 1) Query downstream dependent workflows for the current workflow
-        final WorkflowDefinition upstreamWorkflow = backfillWorkflowDTO.getWorkflowDefinition();
-        final long upstreamWorkflowCode = upstreamWorkflow.getCode();
-
-        List<DependentWorkflowDefinition> downstreamDefinitions =
-                workflowLineageService.queryDownstreamDependentWorkflowDefinitions(upstreamWorkflowCode);
-
-        if (downstreamDefinitions == null || downstreamDefinitions.isEmpty()) {
-            log.info("No downstream dependent workflows found for workflow code {}", upstreamWorkflowCode);
-            return;
+        final boolean isRootCall = BACKFILL_VISITING_WORKFLOWS.get() == null;
+        if (isRootCall) {
+            final Set<Long> visitedCodes = new HashSet<>();
+            visitedCodes.add(backfillWorkflowDTO.getWorkflowDefinition().getCode());
+            BACKFILL_VISITING_WORKFLOWS.set(visitedCodes);
         }
+        try {
+            final Set<Long> visitedCodes = BACKFILL_VISITING_WORKFLOWS.get();
 
-        // 2) Convert upstream backfill time from string to ZonedDateTime as the base business dates for downstream
-        // backfill
-        final List<ZonedDateTime> upstreamBackfillDates = backfillTimeList.stream()
-                .map(DateUtils::stringToZoneDateTime)
-                .collect(Collectors.toList());
+            // 1) Query downstream dependent workflows for the current workflow
+            final WorkflowDefinition upstreamWorkflow = backfillWorkflowDTO.getWorkflowDefinition();
+            final long upstreamWorkflowCode = upstreamWorkflow.getCode();
 
-        // 3) Iterate downstream workflows and build/trigger corresponding BackfillWorkflowDTO
-        for (DependentWorkflowDefinition dependentWorkflowDefinition : downstreamDefinitions) {
-            long downstreamCode = dependentWorkflowDefinition.getWorkflowDefinitionCode();
+            List<DependentWorkflowDefinition> downstreamDefinitions =
+                    workflowLineageService.queryDownstreamDependentWorkflowDefinitions(upstreamWorkflowCode);
 
-            // Prevent self dependency
-            if (downstreamCode == upstreamWorkflowCode) {
-                log.warn("Skip self dependent workflow {}", downstreamCode);
-                continue;
+            if (downstreamDefinitions == null || downstreamDefinitions.isEmpty()) {
+                log.info("No downstream dependent workflows found for workflow code {}", upstreamWorkflowCode);
+                return;
             }
 
-            WorkflowDefinition downstreamWorkflow =
-                    workflowDefinitionDao.queryByCode(downstreamCode).orElse(null);
-            if (downstreamWorkflow == null) {
-                log.warn("Skip dependent workflow {}, definition not found", downstreamCode);
-                continue;
+            // 2) Convert upstream backfill time from string to ZonedDateTime as the base business dates for downstream
+            // backfill
+            final List<ZonedDateTime> upstreamBackfillDates = backfillTimeList.stream()
+                    .map(DateUtils::stringToZoneDateTime)
+                    .collect(Collectors.toList());
+
+            // 3) Iterate downstream workflows and build/trigger corresponding BackfillWorkflowDTO
+            for (DependentWorkflowDefinition dependentWorkflowDefinition : downstreamDefinitions) {
+                long downstreamCode = dependentWorkflowDefinition.getWorkflowDefinitionCode();
+
+                // Prevent self-dependency and circular dependency chains
+                if (visitedCodes.contains(downstreamCode)) {
+                    log.warn("Skip circular dependent workflow {}", downstreamCode);
+                    continue;
+                }
+
+                WorkflowDefinition downstreamWorkflow =
+                        workflowDefinitionDao.queryByCode(downstreamCode).orElse(null);
+                if (downstreamWorkflow == null) {
+                    log.warn("Skip dependent workflow {}, definition not found", downstreamCode);
+                    continue;
+                }
+
+                if (downstreamWorkflow.getReleaseState() != ReleaseState.ONLINE) {
+                    log.warn("Skip dependent workflow {}, release state is not ONLINE", downstreamCode);
+                    continue;
+                }
+
+                // Currently, reuse the same business date list as upstream for downstream backfill;
+                // later we can refine the dates based on dependency cycle configuration in dependentWorkflowDefinition
+                // (taskParams).
+                BackfillWorkflowDTO.BackfillParamsDTO originalParams = backfillWorkflowDTO.getBackfillParams();
+                boolean allLevelDependent = originalParams.isAllLevelDependent();
+                ComplementDependentMode downstreamDependentMode =
+                        allLevelDependent ? originalParams.getBackfillDependentMode() : ComplementDependentMode.OFF_MODE;
+
+                BackfillWorkflowDTO.BackfillParamsDTO dependentParams = BackfillWorkflowDTO.BackfillParamsDTO.builder()
+                        .runMode(originalParams.getRunMode())
+                        .backfillDateList(upstreamBackfillDates)
+                        .expectedParallelismNumber(originalParams.getExpectedParallelismNumber())
+                        // Control whether downstream will continue triggering its own dependencies based on
+                        // allLevelDependent flag
+                        .backfillDependentMode(downstreamDependentMode)
+                        .allLevelDependent(allLevelDependent)
+                        .executionOrder(originalParams.getExecutionOrder())
+                        .build();
+
+                BackfillWorkflowDTO dependentBackfillDTO = BackfillWorkflowDTO.builder()
+                        .loginUser(backfillWorkflowDTO.getLoginUser())
+                        .workflowDefinition(downstreamWorkflow)
+                        .startNodes(null)
+                        .failureStrategy(backfillWorkflowDTO.getFailureStrategy())
+                        .taskDependType(backfillWorkflowDTO.getTaskDependType())
+                        .execType(backfillWorkflowDTO.getExecType())
+                        .warningType(backfillWorkflowDTO.getWarningType())
+                        .warningGroupId(downstreamWorkflow.getWarningGroupId())
+                        .runMode(dependentParams.getRunMode())
+                        .workflowInstancePriority(backfillWorkflowDTO.getWorkflowInstancePriority())
+                        .workerGroup(backfillWorkflowDTO.getWorkerGroup())
+                        .tenantCode(backfillWorkflowDTO.getTenantCode())
+                        .environmentCode(backfillWorkflowDTO.getEnvironmentCode())
+                        .startParamList(backfillWorkflowDTO.getStartParamList())
+                        .dryRun(backfillWorkflowDTO.getDryRun())
+                        .backfillParams(dependentParams)
+                        .build();
+
+                log.info("Trigger dependent workflow {} for upstream workflow {} with backfill dates {}",
+                        downstreamCode, upstreamWorkflowCode, backfillTimeList);
+
+                // 4) Mark as visiting before recursive trigger to detect cycles
+                visitedCodes.add(downstreamCode);
+                execute(dependentBackfillDTO);
             }
-
-            if (downstreamWorkflow.getReleaseState() != ReleaseState.ONLINE) {
-                log.warn("Skip dependent workflow {}, release state is not ONLINE", downstreamCode);
-                continue;
+        } finally {
+            if (isRootCall) {
+                BACKFILL_VISITING_WORKFLOWS.remove();
             }
-
-            // Currently, reuse the same business date list as upstream for downstream backfill;
-            // later we can refine the dates based on dependency cycle configuration in dependentWorkflowDefinition
-            // (taskParams).
-            BackfillWorkflowDTO.BackfillParamsDTO originalParams = backfillWorkflowDTO.getBackfillParams();
-            boolean allLevelDependent = originalParams.isAllLevelDependent();
-            ComplementDependentMode downstreamDependentMode =
-                    allLevelDependent ? originalParams.getBackfillDependentMode() : ComplementDependentMode.OFF_MODE;
-
-            BackfillWorkflowDTO.BackfillParamsDTO dependentParams = BackfillWorkflowDTO.BackfillParamsDTO.builder()
-                    .runMode(originalParams.getRunMode())
-                    .backfillDateList(upstreamBackfillDates)
-                    .expectedParallelismNumber(originalParams.getExpectedParallelismNumber())
-                    // Control whether downstream will continue triggering its own dependencies based on
-                    // allLevelDependent flag
-                    .backfillDependentMode(downstreamDependentMode)
-                    .allLevelDependent(allLevelDependent)
-                    .executionOrder(originalParams.getExecutionOrder())
-                    .build();
-
-            BackfillWorkflowDTO dependentBackfillDTO = BackfillWorkflowDTO.builder()
-                    .loginUser(backfillWorkflowDTO.getLoginUser())
-                    .workflowDefinition(downstreamWorkflow)
-                    .startNodes(backfillWorkflowDTO.getStartNodes())
-                    .failureStrategy(backfillWorkflowDTO.getFailureStrategy())
-                    .taskDependType(backfillWorkflowDTO.getTaskDependType())
-                    .execType(backfillWorkflowDTO.getExecType())
-                    .warningType(backfillWorkflowDTO.getWarningType())
-                    .warningGroupId(downstreamWorkflow.getWarningGroupId())
-                    .runMode(dependentParams.getRunMode())
-                    .workflowInstancePriority(backfillWorkflowDTO.getWorkflowInstancePriority())
-                    .workerGroup(backfillWorkflowDTO.getWorkerGroup())
-                    .tenantCode(backfillWorkflowDTO.getTenantCode())
-                    .environmentCode(backfillWorkflowDTO.getEnvironmentCode())
-                    .startParamList(backfillWorkflowDTO.getStartParamList())
-                    .dryRun(backfillWorkflowDTO.getDryRun())
-                    .backfillParams(dependentParams)
-                    .build();
-
-            log.info("Trigger dependent workflow {} for upstream workflow {} with backfill dates {}",
-                    downstreamCode, upstreamWorkflowCode, backfillTimeList);
-
-            // 4) Recursively call execute to trigger downstream backfill
-            execute(dependentBackfillDTO);
         }
     }
 }

dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/executor/workflow/BackfillWorkflowExecutorDelegateTest.java

@@ -169,6 +169,7 @@ public void testDoBackfillDependentWorkflow_WithDownstream_AllLevelDependent() t
         Assertions.assertEquals(ComplementDependentMode.ALL_DEPENDENT, capturedParams.getBackfillDependentMode());
         Assertions.assertTrue(capturedParams.isAllLevelDependent());
         Assertions.assertEquals(backfillTimeList.size(), capturedParams.getBackfillDateList().size());
+        Assertions.assertNull(captured.getStartNodes());
     }
 
     @Test
@@ -232,5 +233,83 @@ public void testDoBackfillDependentWorkflow_WithDownstream_SingleLevelDependent(
         Assertions.assertEquals(ComplementDependentMode.OFF_MODE, capturedParams.getBackfillDependentMode());
         Assertions.assertFalse(capturedParams.isAllLevelDependent());
         Assertions.assertEquals(backfillTimeList.size(), capturedParams.getBackfillDateList().size());
+        Assertions.assertNull(captured.getStartNodes());
+    }
+
+    @Test
+    public void testDoBackfillDependentWorkflow_SkipWorkflowNotFound() throws Exception {
+        long upstreamCode = 1000L;
+        long downstreamCode = 2000L;
+
+        WorkflowDefinition upstreamWorkflow =
+                WorkflowDefinition.builder().code(upstreamCode).releaseState(ReleaseState.ONLINE).build();
+
+        BackfillWorkflowDTO.BackfillParamsDTO params = BackfillWorkflowDTO.BackfillParamsDTO.builder()
+                .runMode(RunMode.RUN_MODE_SERIAL)
+                .backfillDateList(Collections.<ZonedDateTime>emptyList())
+                .backfillDependentMode(ComplementDependentMode.ALL_DEPENDENT)
+                .allLevelDependent(true)
+                .executionOrder(ExecutionOrder.ASC_ORDER)
+                .build();
+
+        BackfillWorkflowDTO dto = BackfillWorkflowDTO.builder()
+                .workflowDefinition(upstreamWorkflow)
+                .backfillParams(params)
+                .build();
+
+        DependentWorkflowDefinition dep = new DependentWorkflowDefinition();
+        dep.setWorkflowDefinitionCode(downstreamCode);
+
+        when(workflowLineageService.queryDownstreamDependentWorkflowDefinitions(upstreamCode))
+                .thenReturn(Collections.singletonList(dep));
+        when(workflowDefinitionDao.queryByCode(downstreamCode)).thenReturn(Optional.empty());
+
+        Method method = BackfillWorkflowExecutorDelegate.class.getDeclaredMethod(
+                "doBackfillDependentWorkflow", BackfillWorkflowDTO.class, List.class);
+        method.setAccessible(true);
+
+        method.invoke(backfillWorkflowExecutorDelegate, dto, Collections.singletonList("2026-02-01 00:00:00"));
+
+        verify(backfillWorkflowExecutorDelegate, never()).execute(org.mockito.ArgumentMatchers.any());
+    }
+
+    @Test
+    public void testDoBackfillDependentWorkflow_SkipOfflineWorkflow() throws Exception {
+        long upstreamCode = 3000L;
+        long downstreamCode = 4000L;
+
+        WorkflowDefinition upstreamWorkflow =
+                WorkflowDefinition.builder().code(upstreamCode).releaseState(ReleaseState.ONLINE).build();
+
+        WorkflowDefinition offlineDownstream =
+                WorkflowDefinition.builder().code(downstreamCode).releaseState(ReleaseState.OFFLINE).build();
+
+        BackfillWorkflowDTO.BackfillParamsDTO params = BackfillWorkflowDTO.BackfillParamsDTO.builder()
+                .runMode(RunMode.RUN_MODE_SERIAL)
+                .backfillDateList(Collections.<ZonedDateTime>emptyList())
+                .backfillDependentMode(ComplementDependentMode.ALL_DEPENDENT)
+                .allLevelDependent(true)
+                .executionOrder(ExecutionOrder.ASC_ORDER)
+                .build();
+
+        BackfillWorkflowDTO dto = BackfillWorkflowDTO.builder()
+                .workflowDefinition(upstreamWorkflow)
+                .backfillParams(params)
+                .build();
+
+        DependentWorkflowDefinition dep = new DependentWorkflowDefinition();
+        dep.setWorkflowDefinitionCode(downstreamCode);
+
+        when(workflowLineageService.queryDownstreamDependentWorkflowDefinitions(upstreamCode))
+                .thenReturn(Collections.singletonList(dep));
+        when(workflowDefinitionDao.queryByCode(downstreamCode)).thenReturn(Optional.of(offlineDownstream));
+
+        Method method = BackfillWorkflowExecutorDelegate.class.getDeclaredMethod(
+                "doBackfillDependentWorkflow", BackfillWorkflowDTO.class, List.class);
+        method.setAccessible(true);
+
+        method.invoke(backfillWorkflowExecutorDelegate, dto, Collections.singletonList("2026-02-01 00:00:00"));
+
+        verify(backfillWorkflowExecutorDelegate, never()).execute(org.mockito.ArgumentMatchers.any());
     }
 }