[EXPERIMENTAL] feat(dag): migrate DAG support to extension SPI - #611
[EXPERIMENTAL] feat(dag): migrate DAG support to extension SPI#611zhongkechen wants to merge 3 commits into
Conversation
GitHub Actions
Waiting
GitHub Actions
Failure
ecf0c30 to
e25ea9c
Compare
25cf931 to
b08588f
Compare
6a8bbc1 to
818b256
Compare
GitHub Actions
Failure
GitHub Actions
Inactive
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
| while (!inFlight.isEmpty()) { | ||
| var it = inFlight.entrySet().iterator(); | ||
| var entry = it.next(); | ||
| it.remove(); | ||
| String name = entry.getKey(); | ||
| try { | ||
| Object result = entry.getValue().future.get(); |
There was a problem hiding this comment.
Codex AI review
[P1] Completion is observed in launch order. If an earlier task is a long wait/callback while a later task satisfies firstSuccessful or a custom policy, the DAG still blocks on the earlier future. Skips also bypass completion evaluation. Await any settled future and evaluate the policy after every terminal transition, persisting enough state to replay the same completion boundary.
| return switch (kind) { | ||
| case PLAIN -> rehydratePlain(raw, taskName, scope); | ||
| case BATCH -> delegate.deserialize(delegate.serialize(raw), TypeToken.get(MapResult.class)); |
There was a problem hiding this comment.
Codex AI review
[P1] Deserializing raw MapResult.class erases O. A MapResult<MyRecord> is normalized into MapResult<LinkedHashMap>, causing typed access to fail even on first execution. Retain the map output TypeToken and reconstruct a parameterized MapResult or each item with that type; add a POJO replay test.
| } catch (UnrecoverableDurableExecutionException e) { | ||
| throw e; | ||
| } catch (RuntimeException e) { | ||
| results.put(name, failed(name, DagTaskError.of(e))); |
There was a problem hiding this comment.
Codex AI review
[P2] Durable futures may rethrow the original non-RuntimeException Throwable. Such a task currently aborts the DAG after its operation was checkpointed failed, bypassing compensation. Explicitly rethrow SuspendExecutionException and UnrecoverableDurableExecutionException, then catch remaining Throwable as a failed task.
| var cc = unwrap(dcc); | ||
| int succeeded = countByStatus(results, TaskStatus.SUCCEEDED); | ||
| int failed = countByStatus(results, TaskStatus.FAILED); | ||
| if (cc.minSuccessful() != null && succeeded >= cc.minSuccessful()) { | ||
| return DagCompletionReason.MIN_SUCCESSFUL_REACHED; |
There was a problem hiding this comment.
Codex AI review
[P2] When minSuccessful exceeds the registered task count, the threshold can never fire and the scheduler silently returns ALL_COMPLETED or COMPLETED_WITH_FAILURES. Validate this after graph registration and before reserving operations, matching map's existing validation.
| try { | ||
| return delegate.deserialize(delegate.serialize(raw), declared.get()); | ||
| } catch (RuntimeException e) { | ||
| return raw; |
There was a problem hiding this comment.
Codex AI review
[P2] A known declared type must not silently degrade to a raw JSON tree when deserialization fails; this converts a serialization error into a delayed ClassCastException and defeats deserialize-after-serialize validation. Propagate the SerDes failure, reserving raw fallback only for unknown task names.
| // ── invoke ─────────────────────────────────────────────────────────────── | ||
| <T> TaskHandle<T> invoke(String name, String functionName, Class<T> type, DagPayloadFunction payloadFn); | ||
|
|
||
| <T> TaskHandle<T> invoke( | ||
| String name, String functionName, Class<T> type, DagPayloadFunction payloadFn, InvokeConfig config); | ||
|
|
||
| // ── callback ───────────────────────────────────────────────────────────── | ||
| <T> TaskHandle<T> callback(String name, Class<T> type, DagCallbackSubmitter submitter); | ||
|
|
||
| <T> TaskHandle<T> callback( | ||
| String name, Class<T> type, DagCallbackSubmitter submitter, WaitForCallbackConfig config); | ||
|
|
||
| // ── wait ───────────────────────────────────────────────────────────────── | ||
| TaskHandle<Void> wait(String name, Duration duration); | ||
|
|
||
| // ── waitForCondition ────────────────────────────────────────────────────── | ||
| <S> TaskHandle<S> waitForCondition( | ||
| String name, Class<S> type, DagConditionFunction<S> check, WaitForConditionConfig<S> config); | ||
|
|
||
| // ── runInChildContext ───────────────────────────────────────────────────── | ||
| <T> TaskHandle<T> runInChildContext(String name, Class<T> type, DagChildFunction<T> fn); | ||
|
|
||
| <T> TaskHandle<T> runInChildContext(String name, TypeToken<T> type, DagChildFunction<T> fn); | ||
|
|
||
| // ── map ────────────────────────────────────────────────────────────────── | ||
| <I, O> TaskHandle<MapResult<O>> map(String name, Collection<I> items, Class<O> type, MapFunction<I, O> fn); | ||
|
|
||
| <I, O> TaskHandle<MapResult<O>> map( | ||
| String name, Collection<I> items, Class<O> type, MapFunction<I, O> fn, MapConfig config); | ||
|
|
||
| <I, O> TaskHandle<MapResult<O>> map( | ||
| String name, Function<Deps, Collection<I>> items, Class<O> type, MapFunction<I, O> fn); | ||
|
|
||
| <I, O> TaskHandle<MapResult<O>> map( | ||
| String name, Function<Deps, Collection<I>> items, Class<O> type, MapFunction<I, O> fn, MapConfig config); |
There was a problem hiding this comment.
Codex AI review
[P2] Unlike the core APIs, invoke, callback, waitForCondition, and map expose only Class result types, so generic results such as List<MyRecord> cannot be represented and replay as raw collections. Add corresponding TypeToken overloads and delegate the Class overloads to them.
| ## Completion (threshold only in v1) | ||
|
|
||
| `DagConfig.builder().completionConfig(...)` accepts one of six threshold policies: | ||
| `allCompleted`, `allSuccessful`, `firstSuccessful`, `minSuccessful(n)`, `toleratedFailureCount(n)`, | ||
| `toleratedFailurePercentage(p)`. Default (no `completionConfig`) drains the whole reachable graph. `completionReason()` | ||
| reports `ALL_COMPLETED`, `COMPLETED_WITH_FAILURES`, `MIN_SUCCESSFUL_REACHED`, or `FAILURE_TOLERANCE_EXCEEDED`. | ||
|
|
||
| > **v2-deferred:** Custom-predicate (result-based) completion is **not** in v1. `DagCompletionConfig` exposes only the | ||
| > threshold factories, and `DagCompletionReason.CUSTOM_COMPLETION_*` are reserved-but-unreachable. | ||
|
|
||
| ## Results | ||
|
|
||
| `DagResult` provides `getResult(TaskHandle<T>) -> Optional<T>` (typed) and `getResult(String) -> Optional<Object>` | ||
| (untyped), `getStatus(...)`, grouped views (`succeeded()`/`failed()`/`skipped()`), counts, `completionReason()`, and | ||
| `throwIfError()` (throws `DagExecutionException` iff `failureCount() > 0`). |
There was a problem hiding this comment.
Codex AI review
[P2] This page documents a different API: custom completion is public and tested, its completion reasons are reachable, and throwIfError() also throws for CUSTOM_COMPLETION_FAILED. Earlier examples also reference nonexistent getOptional and treat Deps.get as returning T instead of Optional<T>. Rewrite these sections against the actual signatures and behavior so the examples compile.
Codex AI reviewFound blocking completion and serialization issues, plus API/error-handling gaps and inaccurate documentation. Static review only; tests were not run per instruction. Reviewed commit |
1 participant
By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
Warning
Experimental stacked PR. This work depends on draft PR #607 and is not intended for production use. The DAG API, checkpoint shape, and implementation may change before merge or release.
Issue Link, if available
Stacked on #607.
Description
Migrates the experimental DAG implementation to the extension operation SPI introduced by #607.
DurableDagOperation.dag(...)anddagAsync(...)entry points following the newDurable*Operationfacade style.DurableContext.ExtensionContextandExtensionOperation.Durable*Operationclasses and their implementation tests unchanged from the stacked base.Demo/Screenshots
Not applicable. This is an SDK API and execution implementation change.
Checklist
Testing
Unit Tests
Yes. Added focused coverage for
DurableDagOperationand the DAG-internal reserved-context adapter.Integration Tests
Yes. DAG integration and conformance tests were updated for the static operation facade. The complete Maven reactor passes.
Commands run:
Cloud example tests remain disabled by default.
Examples
Existing DAG examples were migrated to
DurableDagOperation; no additional example was required.