maintain trace context after async task

weibocom · Feb 13, 2025 · 574a982 · 574a982
1 parent e904530
commit 574a982
Showing 1 changed file with 9 additions and 3 deletions.
diff --git a/...icene-traversal/src/main/java/com/weibo/rill/flow/olympicene/traversal/DAGOperations.java b/...icene-traversal/src/main/java/com/weibo/rill/flow/olympicene/traversal/DAGOperations.java
@@ -242,8 +242,14 @@ public void finishTaskSync(String executionId, String taskCategory, NotifyInfo n
             }
             if (isTaskCompleted(executionResult)) {
                 timeCheckRunner.remTaskFromTimeoutCheck(executionId, executionResult.getTaskInfo());
-                dagTraversal.submitTraversal(executionId, executionResult.getTaskInfo().getName());
-                invokeTaskCallback(executionId, executionResult.getTaskInfo(), executionResult.getContext());
+                // 在恢复的 context 中触发下一个任务
+                Context parentContext = span != null ? Context.current() : Context.current();
+                runnerExecutor.execute(new ExecutionRunnable(executionId, () -> {
+                    try (Scope ignored = parentContext.makeCurrent()) {
+                        dagTraversal.submitTraversal(executionId, executionResult.getTaskInfo().getName());
+                        invokeTaskCallback(executionId, executionResult.getTaskInfo(), executionResult.getContext());
+                    }
+                }));
             }
             if (StringUtils.isNotBlank(executionResult.getTaskNameNeedToTraversal())) {
                 dagTraversal.submitTraversal(executionId, executionResult.getTaskNameNeedToTraversal());
@@ -257,7 +263,7 @@ public void finishTaskSync(String executionId, String taskCategory, NotifyInfo n
         } finally {
             if (span != null && executionResult.getTaskStatus().isCompleted()) {
                 span.setAttribute("status", executionResult.getTaskStatus().toString());
-                span.end();  // 结束之前保存的 span
+                span.end();
             }
         }
     }