diff --git a/rill-flow-dag/olympicene-traversal/src/main/java/com/weibo/rill/flow/olympicene/traversal/runners/SuspenseTaskRunner.java b/rill-flow-dag/olympicene-traversal/src/main/java/com/weibo/rill/flow/olympicene/traversal/runners/SuspenseTaskRunner.java index 172ca2163..c0c6d9b16 100644 --- a/rill-flow-dag/olympicene-traversal/src/main/java/com/weibo/rill/flow/olympicene/traversal/runners/SuspenseTaskRunner.java +++ b/rill-flow-dag/olympicene-traversal/src/main/java/com/weibo/rill/flow/olympicene/traversal/runners/SuspenseTaskRunner.java @@ -19,6 +19,8 @@ import com.google.common.collect.ImmutableSet; import com.google.common.collect.Maps; import com.google.common.collect.Sets; +import com.weibo.rill.flow.interfaces.model.strategy.Timeline; +import com.weibo.rill.flow.interfaces.model.task.BaseTask; import com.weibo.rill.flow.interfaces.model.task.TaskInfo; import com.weibo.rill.flow.interfaces.model.task.TaskInvokeMsg; import com.weibo.rill.flow.interfaces.model.task.TaskStatus; @@ -31,11 +33,11 @@ import com.weibo.rill.flow.olympicene.core.runtime.DAGInfoStorage; import com.weibo.rill.flow.olympicene.core.runtime.DAGStorageProcedure; import com.weibo.rill.flow.olympicene.core.switcher.SwitcherManager; -import com.weibo.rill.flow.olympicene.traversal.utils.ConditionsUtil; import com.weibo.rill.flow.olympicene.traversal.constant.TraversalErrorCode; import com.weibo.rill.flow.olympicene.traversal.exception.DAGTraversalException; import com.weibo.rill.flow.olympicene.traversal.helper.ContextHelper; import com.weibo.rill.flow.olympicene.traversal.mappings.InputOutputMapping; +import com.weibo.rill.flow.olympicene.traversal.utils.ConditionsUtil; import lombok.extern.slf4j.Slf4j; import org.apache.commons.collections.CollectionUtils; import org.apache.commons.collections.MapUtils; @@ -117,7 +119,25 @@ public ExecutionResult finish(String executionId, NotifyInfo notifyInfo, Map event.eventCode == DAGEvent.TASK_FAILED.getCode() + }) + } + + def "suspense task should be FAILED on timeout when tolerance=false even if skipOnTimeout=true"() { + given: + // tolerance=false,即使 skipOnTimeout=true,超时后也应触发 TASK_FAILED 事件 + String text = "version: 0.0.1\n" + + "namespace: olympicene\n" + + "service: mca\n" + + "name: test\n" + + "type: flow\n" + + "tasks: \n" + + "- category: suspense\n" + + " name: A\n" + + " tolerance: false\n" + + " timeline:\n" + + " timeoutInSeconds: \"120\"\n" + + " skipOnTimeout: \"true\"\n" + + " conditions:\n" + + " - \$.input.[?(@.url == \"bbb\")]\n" + DAG dag = dagParser.parse(text) + + when: + olympicene.submit('timeout_tol_false_1', dag, [:]) + olympicene.wakeup('timeout_tol_false_1', [:], + NotifyInfo.builder() + .taskInfoName('A') + .taskStatus(TaskStatus.FAILED) + .taskInvokeMsg(TaskInvokeMsg.builder().msg("timeout").build()) + .build()) + + then: + // tolerance=false → 触发 TASK_FAILED 回调事件,而不是 TASK_SKIPPED + 1 * callback.onEvent({ + Event event -> event.eventCode == DAGEvent.TASK_FAILED.getCode() + }) + } + def "suspense task should remain FAILED on interruption even if skipOnTimeout=true and tolerance=true"() { + given: + // tolerance=true 且 skipOnTimeout=true,但触发的是打断而非超时,应保持 FAILED + String text = "version: 0.0.1\n" + + "namespace: olympicene\n" + + "service: mca\n" + + "name: test\n" + + "type: flow\n" + + "tasks: \n" + + "- category: suspense\n" + + " name: A\n" + + " tolerance: true\n" + + " timeline:\n" + + " timeoutInSeconds: \"120\"\n" + + " skipOnTimeout: \"true\"\n" + + " inputMappings:\n" + + " - target: \$.input.url\n" + + " source: \$.context.url\n" + + " - target: \$.input.text\n" + + " source: \$.context.text\n" + + " conditions:\n" + + " - \$.input.[?(@.url == \"bbb\")]\n" + + " interruptions:\n" + + " - \$.input.[?(@.text == \"aaa\")]\n" + DAG dag = dagParser.parse(text) + + when: + // 提交 DAG,初始上下文触发 interruption 条件(text=aaa) + olympicene.submit('interrupt_with_skip_flag_1', dag, ["text": "aaa"]) + + then: + // 虽然 tolerance=true && skipOnTimeout=true,但触发的是打断而非超时,应为 TASK_FAILED 而非 TASK_SKIPPED + 1 * callback.onEvent({ + Event event -> event.eventCode == DAGEvent.TASK_FAILED.getCode() + }) + } } \ No newline at end of file diff --git a/rill-flow-interfaces/src/main/java/com/weibo/rill/flow/interfaces/model/strategy/Timeline.java b/rill-flow-interfaces/src/main/java/com/weibo/rill/flow/interfaces/model/strategy/Timeline.java index 20fe95540..720ef795d 100644 --- a/rill-flow-interfaces/src/main/java/com/weibo/rill/flow/interfaces/model/strategy/Timeline.java +++ b/rill-flow-interfaces/src/main/java/com/weibo/rill/flow/interfaces/model/strategy/Timeline.java @@ -26,15 +26,19 @@ @Getter public class Timeline { private String timeoutInSeconds; + private String skipOnTimeout; private String suspenseIntervalSeconds; private String suspenseTimestamp; @JsonCreator public Timeline(@JsonProperty("timeoutInSeconds") String timeoutInSeconds, + @JsonProperty("skipOnTimeout") String skipOnTimeout, @JsonProperty("suspenseIntervalSeconds") String suspenseIntervalSeconds, @JsonProperty("suspenseTimestamp") String suspenseTimestamp) { this.timeoutInSeconds = timeoutInSeconds; + this.skipOnTimeout = skipOnTimeout; this.suspenseIntervalSeconds = suspenseIntervalSeconds; this.suspenseTimestamp = suspenseTimestamp; } } +