Repository navigation
[Fix][Zeta] Resolve transform dependencies without hangs or dropped inputs - #12079
goutamadwant wants to merge 21 commits into
Conversation
DanielLeens
left a comment
There was a problem hiding this comment.
What Problem Does This PR Solve?
The legacy transform-dependency resolver in MultipleTableJobConfigParser#parseTransforms (and its near-duplicate in DryRunConnectValidator) used a Queue<Config> retry loop: poll a transform config, and if its declared inputs are not yet available, push it back to the tail of the queue and try the next one. Two real defects follow from that design:
- If a transform's
plugin_inputcan never be satisfied (typo, forward reference that never resolves, or a genuine dependency cycle), the queue never empties andparseTransforms/--dry-run connectloops forever (the "hangs" in the title). - A multi-input transform was considered "ready" as soon as any of its inputs matched an existing table — the remaining, still-unresolved input ids were silently dropped from the generated action's input set instead of being waited for, so the produced DAG silently missed edges.
This PR replaces the retry queue with an indexed, Kahn-style ready-scheduler (scheduleTransforms) that resolves transforms only once every configured input id is available, detects cycles/unresolvable graphs in linear/O((V+E) log V) time instead of looping, and (for the engine-core path) reproduces the exact legacy poll-count so that default action names (Transform[<index>]-<plugin>) are unchanged for every DAG that previously resolved correctly.
1. Code Change Review
1.1 Core Logic Analysis
Before (MultipleTableJobConfigParser, dev HEAD):
Queue<Config> configList = new LinkedList<>(transformConfigs);
int index = 0;
while (!configList.isEmpty()) {
parseTransform(index++, configList, classLoader, tableWithActionMap, usedTransformNames);
}
...
List<Tuple2<CatalogTable, Action>> inputs =
inputIds.stream().map(tableWithActionMap::get).filter(Objects::nonNull)
.flatMap(Collection::stream).collect(Collectors.toList());
if (inputs.isEmpty()) {
if (transforms.isEmpty()) {
inputs = findLast(tableWithActionMap); // silent "just chain to the last table" fallback
} else {
transforms.offer(config); // requeue forever if never satisfied
return;
}
}After: scheduleTransforms() builds, for every transform, its exact set of input ids and a waitingByInputId reverse index. A TreeSet<Integer> of "ready" indexes is drained by repeatedly picking ceiling(queueHeadIndex) (wrapping to first()), which — see below — is a provably faithful re-encoding of "next item in the rotated retry queue". Every configured input id must be present in tableWithActionMap/availableOutputIds before a transform is scheduled (missingInputIds must be empty); anything left unresolved after the ready-set drains is either the legacy single-transform/empty-input fallback or a hard JobDefineCheckException.
Key Findings
- Correctness of the legacy-order emulation is real, not decorative. I traced the round-robin retry queue by hand: because unready items are always moved to the tail in the order they were polled, the "remaining queue" at any point in time is exactly a rotation of the original index order starting at the position right after the last successfully-removed item.
queueHeadIndexplays the role of that rotation point, andreadyTransformIndexes.ceiling(queueHeadIndex)/first()reproduces "scan the rotated queue forward for the next ready item" without materializing the queue.nextTransformActionIndex+ the Fenwick tree (createRemainingIndexTree/countRemainingIndexes) then count how many poll calls (ready or not) the legacy loop would have consumed to reach that item — this is exactly the value the legacy code fed intoJobConfigParser.createTransformActionName(index, factoryId). I hand-verified this againsttestSchedulesLongReverseDependencyChainOnceInLegacyEvaluationOrder(512-node fully-reversed chain): the triangular-number action indexes the test asserts (511, 1022, 1532, …) match exactly what the legacyO(n²)retry loop would have produced for that adversarial ordering. - Why this matters for production, not just cosmetics:
ActionStateKey.of(Action action)(seatunnel-engine-server/.../checkpoint/ActionStateKey.java:35) derives the checkpoint state key fromaction.getName(), and unnamed transforms get their name fromcreateTransformActionName(index, factoryId). If this PR had produced a different index/name for an existing valid DAG, then upgrading SeaTunnel and restarting/restoring a running job from a retained checkpoint or savepoint would silently fail to rebind operator state to the correct action (state loss, not a crash). The PR's insistence on bit-for-bit legacy index parity is the right thing to prioritize here, and it is backed by a real (if unusually elaborate) test, not just asserted in the description. - Runtime path:
This is on the normal submission path for every job with a "complex" graph (more than one source/sink or more than one transform) —
MultipleTableJobConfigParser#parse() -> ConfigParserUtil.checkGraph(sources, transforms, sinks) [Phase 6: checkTransformCycles - new] -> parseTransforms(transformConfigs, classLoader, tableWithActionMap) -> scheduleTransforms(transformConfigs, tableWithActionMap.keySet()) [new O((V+E) log V) scheduler] -> parseTransform(actionIndex, config, legacyFallback, ...) [per resolved transform, in legacy order]checkGraphandparseTransformsare called unconditionally fromparse(), including the savepoint-restore branch (isStartWithSavePoint). It is not a boundary-only fix. - Genuine (disclosed) behavior change: for a "complex" graph with 2+ transforms, the old code's
findLast()fallback could trigger for any transform that happened to be the last item remaining in the retry queue when its input never resolved — not just for genuinely single-transform jobs. That means a mis-configured multi-transform job (typo'dplugin_input) could previously submit successfully while silently chaining from the wrong upstream table.testDoesNotApplyLegacyFallbackAfterAnotherTransformWasScheduledconfirms the new code now throwsJobDefineCheckExceptionin that case instead. This is the correct fix (silent misrouting of data is worse than a loud submission-time error), and the PR explicitly discloses it in "Does this PR introduce any user-facing change?" — but see Issue 1 below regardingincompatible-changes.md. - The
Set<URL> jarUrls = new HashSet<>(getTransformPluginJarPaths(config));line was moved to after the input-availability check (comment: "Plugin discovery only happens after all configured dependencies are available"). This avoids repeated/unnecessary classloader jar discovery for transforms that used to be re-polled many times; also avoids doing plugin discovery for a config that will ultimately be rejected as unresolvable. checkTransformCycles(newConfigParserUtilPhase 6) is a textbook Kahn's-algorithm cycle check plus a linear-time concrete-cycle-path reconstruction (findTransformCycle) for the error message. I confirmed it is only invoked for "complex" graphs (checkComplexGraph), never forisSimpleGraph(≤1 source/sink/transform) — this preserves the historical "single-transform self-reference silently tolerated" behavior verified bytestPreservesSimpleGraphCompatibility, while adding real cycle rejection everywhere a cycle was actually reachable before (previously the oldcheckInputId/checkLinkedphases could not detect a self-loop at all, since a transform's own freshly-created vertex satisfies its own input check).
1.2 Compatibility Impact
Partially incompatible, and disclosed. No config option, default, API, or serialization format is added or removed. However:
- A previously-succeeding multi-transform job whose
plugin_inputreferenced a non-existent/unresolvable id (and thus silently fell back tofindLast()) will now fail fast at parse time withJobDefineCheckException/ConfigCheckException. The PR body honestly states this ("Yes" to user-facing change) but the checklist marksincompatible-changes.mdas "not required." I'd treat this as a genuine, if narrow, incompatibility (see Issue 1). - Cyclic transform graphs (which previously hung indefinitely) now fail fast — this is strictly a fix, not a regression, since no valid job could ever have completed from a true cycle.
- Legacy default action naming/indexing for every currently-valid DAG is preserved bit-for-bit (see 1.1), which is the critical compatibility property for checkpoint/savepoint restore across an upgrade.
1.3 Performance / Side-Effect Analysis
- Old: unbounded (hangs) on cyclic/unresolvable graphs;
O(n²)worst case (e.g., a fully-reverse-declared chain) on valid graphs, plus repeated jar/classloader discovery for every failed poll. - New:
O((V+E) log V)(TreeSet operations) for scheduling,O(V+E)memory, linear cycle detection, and jar discovery deferred to once-per-successfully-scheduled-transform. This is a straightforward win with no new unbounded structures, no new threads, no new locks — low risk of side effects for a config-parse-time-only code path (single-threaded, not on the streaming hot path). DryRunConnectValidatorcarries a near-identicalscheduleTransforms/ScheduledTransformimplementation without the Fenwick-tree action-index machinery — correctly recognized as unnecessary there because the dry-run tool's index is purely a display counter, not a persisted state key.
1.4 Error Handling and Logging
- New exceptions (
JobDefineCheckExceptionin engine-core,ConfigCheckExceptionin the dry-run validator) carry the offending transform's output id, its missing input ids, and the currently-available output ids — good, actionable context, consistent with the project's "exceptions must include sufficient context" rule. nextTransformActionIndexexplicitly guards theintnarrowing (Integer.MAX_VALUEoverflow) with a clear message rather than silently wrapping to a negative/incorrect action index — good defensive coding for a value that ultimately becomes part of a checkpoint state key.- No exceptions are swallowed; no sensitive data is logged.
2. Code Quality Assessment
2.1 Coding Standards
- New public/package-private methods (
scheduleTransforms,nextTransformActionIndex, the Fenwick-tree helpers,checkTransformCycles,findTransformCycle) all have explanatory comments describing why, not just what — e.g. the Fenwick-tree comment correctly explains it exists to avoid replaying every unsuccessful poll. This matches the project's comment-density expectations for non-trivial/algorithmic logic. ScheduledTransform(both copies) is a small package-private mutable holder class; fields are undocumented individually (e.g.unresolvedInputCount,legacyFallback,scheduled) but their names are self-explanatory and the class as a whole is used in a very narrow, well-commented scope. Not a blocker.- Minor:
scheduleTransformsandScheduledTransformare duplicated near-verbatim betweenMultipleTableJobConfigParser(engine-core) andDryRunConnectValidator(core-starter). I understand these live in different modules with different dependency graphs (engine-core is not a dependency of seatunnel-starter), so full de-duplication may not be trivial, but this is worth a follow-up ticket — two copies of a subtle scheduling algorithm are two places for a future bug to diverge.
2.2 Test Coverage and Test Stability
Coverage is genuinely thorough for this kind of change: two-node/self cycles, partially/wholly unresolved inputs, mixed source+transform inputs, duplicate input ids, out-of-order valid chains (including the 512-node worst-case reverse chain), the legacy simple-graph and explicit-empty-input fallbacks, and the Integer.MAX_VALUE action-index boundary. All three touched test classes (ConfigParserUtilTest, MultipleTableJobConfigParserTest, SeaTunnelConfValidateCommandTest) are pure, deterministic unit tests operating on in-memory Config objects — no Thread.sleep, no Awaitility, no shared mutable static state driving timing, no container startup. Stability rating: Stable. (The one static field, DryRunTestTransformFactory.CREATED_TABLES, is properly synchronized and reset per test via resetCreatedTables().)
2.3 Documentation Updates
None made; the PR checklist argues docs aren't needed because no config/API surface changed. I agree for the cycle-detection/hang fix itself. I partially disagree for the "multi-transform unresolved input now throws instead of silently falling back" behavior change — see Issue 1.
3. Architectural Soundness
3.1 Elegance of the Solution
This is a genuine, long-term fix, not a workaround — it replaces an algorithmically unsound retry loop with a proper dependency scheduler and adds the cycle-detection phase that was structurally missing from ConfigParserUtil. The extra complexity (Fenwick tree, legacy-order emulation) is not incidental complexity; it's the necessary cost of fixing the algorithm without breaking the checkpoint-state-key compatibility that depends on the old, algorithmically-poor ordering. That is a defensible trade-off for a project with the compatibility bar this one has.
3.2 Maintainability
The two duplicated scheduleTransforms implementations are the main maintainability wart (see 2.1). Everything else is well-isolated, package-private/static, and covered by focused unit tests that would catch a future regression in the scheduling order.
3.3 Extensibility
The ScheduledTransform record-like holder and the ready-set/waiting-map structure would extend cleanly if a future change needed richer dependency metadata (e.g., per-edge type information) without touching the public parseTransforms entry point.
3.4 Historical-Version Compatibility
This is the strongest part of the PR: default action names/indexes for every previously-resolvable DAG are provably unchanged (traced by hand + backed by the 512-node adversarial test), which is exactly what checkpoint/savepoint state-key rebinding (ActionStateKey) needs across an in-place upgrade. The one behavior change that isn't fully "historical-compatible" is the multi-transform silent-fallback removal, which is an intentional bug fix rather than a compatibility regression, and is disclosed in the PR body.
4. Issue Summary
| # | Issue | Location | Severity |
|---|---|---|---|
| 1 | Disclosed user-facing behavior change (multi-transform unresolved plugin_input now fails fast instead of silently falling back to findLast()) is not reflected in docs/en/introduction/concepts/incompatible-changes.md, despite the PR body itself answering "Yes" to "does this introduce a user-facing change?" |
docs/en/introduction/concepts/incompatible-changes.md (not touched) |
Medium |
| 2 | scheduleTransforms/ScheduledTransform logic is duplicated near-verbatim between MultipleTableJobConfigParser (engine-core) and DryRunConnectValidator (core-starter); a future fix to the scheduling algorithm could easily be applied to only one copy |
seatunnel-engine/seatunnel-engine-core/.../MultipleTableJobConfigParser.java, seatunnel-core/seatunnel-starter/.../DryRunConnectValidator.java |
Low |
| 3 | ScheduledTransform's individual mutable fields (unresolvedInputCount, scheduled, legacyFallback) have no per-field comments; class-level intent is clear from context but a one-line comment per field would help a future reader who lands only on this class |
Both ScheduledTransform inner classes |
Low |
5. Merge Recommendation
Conclusion: Ready to merge after fixes
- Blockers — none from a correctness/compatibility standpoint. I would still like Issue 1 addressed before merge: either add a short
incompatible-changes.mdentry describing the "multi-transform unresolved input now fails at parse time instead of silently mis-wiring" change, or have a maintainer explicitly confirm in the PR thread that a bug-fix removing silent data mis-wiring is intentionally exempted from that doc (I'd be fine with either outcome, I just want it recorded). - Recommended fixes (non-blocking) — consider de-duplicating the two
scheduleTransformscopies in a follow-up (Issue 2); a short comment perScheduledTransformfield would help readability (Issue 3).
This is a well-reasoned fix for a real production hazard (an unbounded retry loop that can hang job submission/dry-run on a cyclic or malformed transform graph) and it goes to unusual lengths — successfully, as far as I can verify by hand-tracing the algorithm and cross-checking against the adversarial 512-node test — to preserve the exact legacy action-naming/indexing that checkpoint and savepoint state restoration depends on. The test suite is deterministic and covers the interesting edge cases (cycles, partial/whole unresolved inputs, duplicate inputs, reverse-order chains, both fallback paths, and the integer overflow boundary). Nice work, and thank you for being upfront about the one behavior change in the PR description rather than glossing over it.
Signed-off-by: goutamadwant <workwithgoutam@gmail.com>
Signed-off-by: goutamadwant <workwithgoutam@gmail.com>
DanielLeens
left a comment
There was a problem hiding this comment.
Thanks for the update, @goutamadwant — I re-reviewed the current head (6d82446fb414fad128a7906082e173b01034d183) in full. The good news: the core algorithm files (ConfigParserUtil.java, MultipleTableJobConfigParser.java, DryRunConnectValidator.java) are byte-for-byte identical to what I already hand-traced in my previous round — I diffed them against the commit I last reviewed and confirmed zero changes. What's new in this head is (a) a merge of upstream dev (bringing in unrelated commits like the Shopify source and Aerospike/AmazonSqs changes — none of that touches this PR's own logic) and (b) a genuine, targeted commit that addresses the one blocker I raised last time.
What Problem Does This PR Solve?
- User pain point: the legacy transform-dependency resolver (
MultipleTableJobConfigParser#parseTransforms, and its near-duplicate inDryRunConnectValidator) used a retry queue that re-polled unready transforms forever. A transform whoseplugin_inputcan never resolve (typo, forward reference, or a real dependency cycle) makes job submission or--dry-run connecthang indefinitely. Separately, a multi-input transform was treated as "ready" the moment any one of its inputs resolved, silently dropping the rest instead of waiting for them — producing a DAG with missing edges. - Fix approach: replace the retry queue with an indexed, Kahn-style ready-scheduler that only resolves a transform once every declared input is available, adds real cycle detection (
checkTransformCycles), and reproduces the legacy poll-count exactly so default action names/indexes are unchanged for every graph that previously resolved correctly. - One-sentence summary: turns an unbounded, occasionally-hanging, silently-lossy retry loop into a proper, linear-time dependency scheduler that fails fast and loud on bad input instead of hanging or mis-wiring data — while deliberately preserving the exact legacy action-naming needed for checkpoint/savepoint compatibility.
1. Code Change Review
1.1 Core Logic Analysis
I re-verified the two load-bearing claims from my previous round myself, against the current head, rather than restating them from memory:
- Normal-path reach, including savepoint restore:
MultipleTableJobConfigParser.java:252callsConfigParserUtil.checkGraph(...)and:281callsparseTransforms(...), both unconditionally inparse(), before theisStartWithSavePoint-gated branch at:257/:446. So this is on the normal submission path for every job with more than one source/sink/transform, restore included — not a boundary-only fix. - Checkpoint/savepoint naming dependency is real:
ActionStateKey.of(Action)(seatunnel-engine-server/.../checkpoint/ActionStateKey.java:41) builds the state key fromaction.getName(), andMultipleTableJobConfigParser.java:535derives that name viaJobConfigParser.createTransformActionName(index, factoryId)— the sameindexthe new scheduler is at pains to reproduce bit-for-bit for previously-valid graphs. If the new scheduler had produced a different index for an existing valid DAG, an in-place upgrade with a retained checkpoint/savepoint would silently fail to rebind operator state to the right action — a state-loss bug, not a crash, and exactly the kind of thing that's easy to miss in review. The PR's insistence on legacy-index parity is the correct priority, and it's backed bytestSchedulesLongReverseDependencyChainOnceInLegacyEvaluationOrder(a 512-node fully-reversed chain whose expected triangular-number action indexes I independently re-derived match what the oldO(n²)retry loop would have produced for that adversarial ordering). - Genuine, disclosed behavior change: previously, a multi-transform job with an unresolvable
plugin_inputcould silently fall back tofindLast()(chaining to the wrong upstream table) instead of failing. The new code throwsJobDefineCheckExceptioninstead (testDoesNotApplyLegacyFallbackAfterAnotherTransformWasScheduled). This is the right fix — silently mis-wiring data is worse than a loud submission-time error — and it's exactly the item my previous round flagged as needing anincompatible-changes.mdentry.
What's new in this head — Issue 1 resolved:
+### Transform Dependency Resolution
+
+- **Behavior change: Reject unresolved inputs in multi-transform jobs**
+ - **Affected components**: Zeta job configuration parsing and `--dry-run connect`.
+ - **Description**: Transforms now wait for all declared `plugin_input` dependencies. A
+ multi-transform configuration with unresolved inputs is rejected instead of silently
+ dropping unavailable inputs or connecting the last unresolved transform to an unintended
+ upstream table. ...
+ - **Migration guide**: Correct `plugin_input` references ... Reordering transforms is not
+ required for an otherwise valid dependency graph. The existing single-transform fallback
+ and terminal explicit-empty-input fallback are retained ... Valid graphs whose dependencies
+ were previously resolved correctly retain their evaluation order and default transform
+ action names. (#12079)added to both docs/en/introduction/concepts/incompatible-changes.md and docs/zh/introduction/concepts/incompatible-changes.md. I read both versions side-by-side; the Chinese text is a faithful translation, and the content accurately reflects what the code actually does (I cross-checked "the existing single-transform fallback ... retained" against checkComplexGraph/isSimpleGraph in ConfigParserUtil.java, and it holds — cycle detection is only applied to complex graphs, so the historical simple-graph self-reference tolerance is preserved as stated).
1.2 Compatibility Impact
Partially incompatible, and now correctly disclosed. No Option, default, API, or serialization format is added or removed. The one real incompatibility — a previously-succeeding multi-transform job with an unresolvable plugin_input now fails fast instead of silently mis-wiring — is exactly the kind of change our project rules require in incompatible-changes.md, and it is now there in both languages with correct migration guidance. Legacy action naming/indexing for every currently-valid DAG is preserved bit-for-bit, which is the property that actually matters for checkpoint/savepoint restore across an upgrade.
1.3 Performance / Side-Effect Analysis
Unchanged from my previous assessment (the algorithm files are identical): old behavior was unbounded on cyclic graphs and O(n²) worst-case on valid ones, with repeated classloader/jar discovery on every failed poll; new behavior is O((V+E) log V) scheduling with jar discovery deferred until a transform is actually ready to be scheduled. Single-threaded, config-parse-time-only code path — not on the streaming hot path, no new locks, no new unbounded structures.
1.4 Error Handling and Logging
Unchanged and still correct: JobDefineCheckException/ConfigCheckException carry the offending output id, missing input ids, and currently-available output ids; nextTransformActionIndex explicitly guards int overflow with a clear message rather than silently wrapping to a wrong action index (which would otherwise corrupt a checkpoint state key). No exceptions swallowed, no sensitive data logged.
No formal blocking issues found this round — see Section 4.
2. Code Quality Assessment
2.1 Coding Standards
Unchanged from my previous review: new methods (scheduleTransforms, nextTransformActionIndex, the Fenwick-tree helpers, checkTransformCycles, findTransformCycle) are well-commented on why, matching the project's expectations for non-trivial algorithmic code.
2.2 Test Coverage and Test Stability
Coverage is unchanged and remains genuinely thorough: cycles (including self-loops), partially/wholly unresolved inputs, duplicate input ids, out-of-order valid chains including the 512-node adversarial reverse chain, both legacy fallback paths, and the Integer.MAX_VALUE action-index boundary, across ConfigParserUtilTest, MultipleTableJobConfigParserTest, and SeaTunnelConfValidateCommandTest. All are pure, deterministic in-memory unit tests — no Thread.sleep, no Awaitility, no container startup, no shared timing-sensitive static state (the one static field, DryRunTestTransformFactory.CREATED_TABLES, is reset per test via resetCreatedTables()).
Stability rating: Stable. No new test code was added in this round beyond what I already rated Stable previously; this conclusion is reconfirmed against the unchanged test files in the current head.
2.3 Documentation Updates
Resolved. docs/en and docs/zh incompatible-changes.md entries now cover the one disclosed behavior change; no other user-facing config/API surface changed, so no further documentation is required.
3. Architectural Soundness
3.1 Elegance of the Solution
Long-term solution, not a workaround — it replaces an algorithmically unsound retry loop with a proper dependency scheduler and adds cycle detection that was structurally missing, while preserving the checkpoint-relevant legacy ordering. That combination is a defensible, deliberate trade-off, not incidental complexity.
3.2 Maintainability
The two near-duplicate scheduleTransforms/ScheduledTransform implementations (engine-core vs. core-starter) remain the main maintainability wart, as noted previously — still non-blocking (Issue 2 below), since the two modules genuinely don't share a dependency edge that would make de-duplication trivial.
3.3 Extensibility
Unchanged assessment: the ready-set/waiting-map structure extends cleanly if richer per-edge dependency metadata is needed later.
3.4 Historical-Version Compatibility
This remains the strongest part of the PR: legacy action names/indexes for every previously-resolvable DAG are provably unchanged (hand-traced + backed by the 512-node test), which is exactly what ActionStateKey-based checkpoint/savepoint rebinding needs across an in-place upgrade. The one non-"historical-compatible" behavior change (multi-transform silent-fallback removal) is an intentional, now-documented bug fix.
4. Issue Summary
| # | Issue | Location | Severity |
|---|---|---|---|
| 1 | (Resolved in current head) Disclosed user-facing behavior change was not reflected in incompatible-changes.md in the previously-reviewed commit — now added in both docs/en and docs/zh. Recording as resolved, not re-opening as a new issue. |
docs/en/introduction/concepts/incompatible-changes.md, docs/zh/introduction/concepts/incompatible-changes.md |
Resolved |
| 2 | scheduleTransforms/ScheduledTransform logic is duplicated near-verbatim between MultipleTableJobConfigParser (engine-core) and DryRunConnectValidator (core-starter); a future algorithm fix could be applied to only one copy. |
seatunnel-engine/seatunnel-engine-core/.../MultipleTableJobConfigParser.java, seatunnel-core/seatunnel-starter/.../DryRunConnectValidator.java |
Low |
| 3 | ScheduledTransform's mutable fields (unresolvedInputCount, scheduled, legacyFallback) have no per-field comments; a one-line comment per field would help a reader who lands only on this class. |
Both ScheduledTransform inner classes |
Low |
No High or Medium severity issues remain open. Issues 2 and 3 are unchanged, non-blocking suggestions carried over from my previous round.
5. Merge Recommendation
Conclusion: Ready to merge
- Blockers — must be fixed: none. The Medium-severity documentation gap from my previous round is resolved in the current head.
- Recommended fixes — non-blocking: consider de-duplicating the two
scheduleTransformscopies in a follow-up (Issue 2); a short comment perScheduledTransformfield would help readability (Issue 3). Neither affects correctness or compatibility.
Overall assessment: this remains a well-reasoned, root-cause fix for a real production hazard — an unbounded retry loop that can hang job submission/dry-run on a cyclic or malformed transform graph — and it goes to real, verified lengths to preserve the exact legacy action-naming/indexing that checkpoint and savepoint restoration depends on across an upgrade. The one blocker from my last round (missing incompatible-changes.md entry for a disclosed behavior change) is now fixed correctly in both languages, and I independently re-verified the code itself is unchanged and the fix still holds. The current head's Build check is still queued on the fork at the time of this review — I'd like to see that come back green, but there's no code-side blocker left from my side. Thanks for tracking down and fixing a genuinely subtle correctness issue, and for taking the documentation feedback seriously.
DanielLeens
left a comment
There was a problem hiding this comment.
Follow-up: the fork Build run for this head (6d82446) that was queued at the time of my last review has since completed successfully -- all required jobs are green. No code-side blocker remains from my last review. This looks ready for a maintainer's final merge action.
|
@nzw921rx @davidzollo CI is passing on the latest head, and the blocking review feedback is resolved. Would you mind reviewing the transform dependency fix and merging it if you are happy with the changes? |
|
@goutamadwant Confirming nothing has changed on my side since the follow-up above: CI is still green on The merge action itself isn't mine to take though -- my account here is review/comment-only, not a committer with merge rights, so it's @nzw921rx or @davidzollo who need to make the final call. From a code-correctness standpoint I have no objection to them merging as-is; ping me again if a new commit lands and I'll gladly take another pass. |
Signed-off-by: goutamadwant <workwithgoutam@gmail.com>
|
@DanielLeens Update in 2af0068: Synced dev and resolved the documentation conflicts, retaining both the transform compatibility note and upstream notes. All 66 focused core, client and configuration-validation tests pass on Java 11. No additional resolver logic change was needed. Fresh CI has been triggered. |
DanielLeens
left a comment
There was a problem hiding this comment.
LGTM re-confirmed. I re-ran the full review from scratch on the current head (2af00684c) rather than trusting my two earlier passes, and the outcome is unchanged: no code-side blocker.
What Problem Does This PR Solve?
The legacy transform-dependency resolver (MultipleTableJobConfigParser#parseTransforms, and its near-duplicate in DryRunConnectValidator) used a Queue<Config> retry loop: poll a transform, and if its plugin_input isn't ready yet, push it back to the tail and try the next one. Two real defects followed: (1) a transform whose input can never resolve (typo, forward reference, real cycle) makes the queue spin forever, hanging job submission or --dry-run connect; (2) a multi-input transform was treated as "ready" the moment any one of its inputs resolved, silently dropping the rest instead of waiting, producing a DAG with missing edges.
The fix replaces the retry queue with an indexed, Kahn-style ready-scheduler (scheduleTransforms) that only resolves a transform once every declared input is available, adds explicit cycle detection (ConfigParserUtil#checkTransformCycles), and reproduces the legacy retry-queue's exact poll-count via a Fenwick tree so that default action names/indexes (Transform[<index>]-<plugin>) are byte-identical for every graph that previously resolved correctly — preserving checkpoint/savepoint action-name compatibility.
One-sentence summary: turns an unbounded, occasionally-hanging, silently-lossy retry loop into a linear-time dependency scheduler that fails fast and loud on bad configuration instead of hanging or mis-wiring the DAG.
Before / After Example
Config with t1 declared before its actual dependency t2, and t2 accidentally missing plugin_output:
transform {
sql { plugin_input = [t2], plugin_output = t1 }
sql { plugin_input = [t1], plugin_output = t2 } # cycle: t1 <-> t2
}
- Before:
MultipleTableJobConfigParserpollst1(needst2, not ready, re-offer), pollst2(needst1, not ready, re-offer) — forever.seatunnel.shnever returns; the job never starts and never fails. - After:
checkGraph's Phase 6 (checkTransformCycles) detects the cycle before scheduling even starts and throwsJobDefineCheckException: Transform dependency cycle detected: t1 -> t2 -> t1. Check 'plugin_input' and 'plugin_output' options.immediately.
Runtime Flow That Reaches The Changed Code
seatunnel.sh submit / --dry-run connect
-> MultipleTableJobConfigParser.parse() / DryRunConnectValidator.validate()
-> ConfigParserUtil.checkGraph(sources, transforms, sinks) [Phase 1-5 existing, Phase 6 NEW: checkTransformCycles]
-> parseSource(...) for each source (populates tableWithActionMap)
-> parseTransforms(transformConfigs, ...)
-> MultipleTableJobConfigParser.scheduleTransforms(...) NEW: Kahn's-algorithm ready queue
-> parseTransform(actionIndex, config, legacyFallback, ...) for each, in resolved order
-> parseSink(...) for each sink
This is the job-definition/dry-run path, executed once per job submission on the client/master side — not a per-record hot path.
1. Code Change Review
1.1 Core Logic Analysis
I re-diffed this exact head against the head I last reviewed and approved-in-substance (6d82446fb414, my comment from 09-05/09-11): zero bytes changed in any of the actual logic/test files (ConfigParserUtil.java, MultipleTableJobConfigParser.java, DryRunConnectValidator.java, DryRunTestSinkFactory.java, DryRunTestTransformFactory.java, all four test files). The only diff in this round is a dev-sync merge that touched only docs/en(zh)/introduction/concepts/incompatible-changes.md, and it's a clean merge — I checked for leftover conflict markers (none) and confirmed this PR's own compatibility entry (... (#12079)) is verbatim-preserved alongside the unrelated entries dev picked up in between (REST pagination validation, JDBC XA restore, CDC stale-table pruning, etc. — none overlapping this PR's files).
Since the code itself is unchanged, I re-traced the algorithm fresh rather than rubber-stamping my prior conclusion:
scheduleTransforms (MultipleTableJobConfigParser.java:578-632): builds waitingByInputId (input id -> dependent transforms) and seeds readyTransformIndexes with any transform whose inputs are already fully available. It only marks a transform ready when unresolvedInputCount hits 0 — i.e. every declared input, not just one — which directly fixes defect #2. The queueHeadIndex/ceiling/wrap-to-first() selection order reproduces the legacy circular-queue's evaluation order (not just which transforms get scheduled, but in what order), and nextTransformActionIndex + the Fenwick tree (createRemainingIndexTree/updateRemainingIndexTree/countRemainingIndexes) reproduces the legacy action-index numbering by counting how many still-unresolved indexes the old retry-queue would have cycled through before reaching the same transform — without literally replaying O(n^2) failed polls.
I verified this claim against ConfigParserUtilTest#testSchedulesLongReverseDependencyChainOnceInLegacyEvaluationOrder, which builds the worst case for the old algorithm (512 transforms declared in fully-reversed dependency order) and asserts the exact action index per transform via the closed-form expectedActionIndex += transformCount - evaluationIndex. Hand-tracing the first two terms against the old poll/offer semantics (first resolvable transform is the last item in the initial queue, requiring a full cycle through all 511 others before that queue position is reached) confirms the formula matches the old algorithm's behavior, and the test encodes it as a real byte-identical assertion rather than "runs without throwing."
Key finding (Low, new observation from this round): DryRunConnectValidator.scheduleTransforms does not call ConfigParserUtil.checkGraph, so --dry-run connect never runs checkTransformCycles. A cyclic transform config in dry-run mode is still correctly rejected (it falls into scheduleTransforms's generic "Unable to resolve transform dependencies: [t1 <- [t2], t2 <- [t1]]..." branch, not an infinite loop), but the message doesn't call it a "cycle" or format it as a path (t1 -> t2 -> t1) the way the real submission path now does. Not a correctness bug — both paths correctly fail fast — just a minor diagnostic-quality inconsistency between the two near-duplicate resolvers. See Issue 2.
1.2 Compatibility Impact
Fully compatible for every graph that previously resolved successfully: action indexes/default names are preserved byte-for-byte (proven by the reverse-chain test), and the existing single-transform legacy fallback (findLast) and explicit-empty-plugin_input fallback are both preserved as documented in incompatible-changes.md. For graphs that previously appeared to work by accident (multi-input transform silently missing an edge) or previously hung forever (cycle), behavior now changes to fail-fast at submission time — this is exactly the trade-off documented in both docs/en and docs/zh incompatible-changes.md, worded consistently in both languages.
1.3 Performance / Side-Effect Analysis
This runs once per job submission (client/master side), not per-record, so raw CPU cost is a non-issue either way. The new algorithm is O((V+E) log V) (TreeSet ceiling/first) versus the old O(n^2) worst case (or O(infinity) for a hang) — strictly better. Memory: O(V+E) for waitingByInputId plus O(V) for the Fenwick tree, trivially small for realistic transform counts. No new threads, no I/O, no lock contention — single-threaded parsing code, unchanged concurrency model.
1.4 Error Handling and Logging
All failure modes throw a typed exception (JobDefineCheckException / ConfigCheckException) with the concrete unresolved transform id(s), their missing input ids, and the set of available output ids — a real improvement over the old behavior of silently continuing or hanging with no diagnostic at all. No swallowed exceptions, no sensitive data logged, log.debug used appropriately for phase tracing.
2. Code Quality Assessment
2.1 Coding Standards
Issue 1 (Medium): The new package-private static methods scheduleTransforms, nextTransformActionIndex, checkTransformCycles, findTransformCycle, and the new nested class ScheduledTransform (in both MultipleTableJobConfigParser and DryRunConnectValidator) implement genuinely non-trivial algorithms (topological scheduling with legacy-index reproduction via a Fenwick tree; cycle detection via path-indexed DFS) but have only inline // comments inside the method bodies, not a Javadoc header on the method/class signature. Per project convention, complex logic like this should carry a short Javadoc explaining the contract (inputs, what "legacy fallback" means, why the Fenwick tree is needed) so a future maintainer touching parseTransform's call site doesn't have to re-derive the algorithm from the inline comments alone. Non-blocking, but worth a follow-up.
2.2 Test Coverage and Test Stability
Coverage is strong and specifically targets the two original defects plus the edge cases a scheduler like this tends to get wrong: cycle detection (testRejectsCyclicTransformDependencies, testRejectsSelfCycleInComplexGraph), out-of-order declaration (testAllowsTransformsDeclaredOutOfDependencyOrder), the worst-case legacy-index reproduction (testSchedulesLongReverseDependencyChainOnceInLegacyEvaluationOrder, 512 transforms), integer-overflow guarding (testRejectsLegacyActionIndexOutsideIntegerRange), the "don't apply legacy fallback once another transform has already scheduled" guard, and ambiguous multiple-empty-input rejection. At the MultipleTableJobConfigParser integration level, testParseTransformsRejectsPartiallyResolvedInputsBeforePluginDiscovery specifically proves plugin-jar discovery is deferred until after dependency resolution (closing a subtler resource-cost angle of the original bug).
Rating: Stable. All new tests operate on in-memory Config/ScheduledTransform objects with deterministic TreeSet/LinkedHashMap iteration order, no Thread.sleep, no clock/network/container dependency, no shared static state between tests.
2.3 Documentation Updates
docs/en and docs/zh incompatible-changes.md both carry a matching, accurate entry referencing #12079; content is consistent between languages and correctly scoped (only genuinely-changed graphs are called out, valid graphs are explicitly said to be unaffected).
3. Architectural Soundness
3.1 Elegance of the Solution
Precise fix. The Fenwick-tree trick to reproduce a legacy O(n^2) algorithm's numbering in better-than-quadratic time, backed by a hand-verifiable closed-form test, is a genuinely elegant way to satisfy "fix the bug" and "don't break checkpoint/savepoint action naming" simultaneously without literally keeping the old slow path around.
3.2 Maintainability
Good, modulo the Javadoc gap in Issue 1. The duplication between MultipleTableJobConfigParser.scheduleTransforms and DryRunConnectValidator.scheduleTransforms (near-identical Kahn's-algorithm bodies, one with the Fenwick tree for index fidelity, one without since dry-run output ordering isn't a compatibility concern) is pre-existing duplication (the two classes were already independent near-duplicates before this PR) rather than something this PR introduces; not asking for a consolidation here, but worth a mental note for anyone touching either copy in the future to check the other.
3.3 Extensibility
Fine — ScheduledTransform cleanly separates "what was configured" from "where it landed in the schedule," which would make it straightforward to add e.g. parallel-safe scheduling metadata later without touching the core algorithm.
3.4 Historical-Version Compatibility
No config/protocol/serialization format changes. Action-name compatibility for checkpoint/savepoint restore across an upgrade is the specific property this PR set out to preserve, and it's the one property backed by a dedicated worst-case test — appropriately, since this is the one place an accidental behavior change would be silently destructive (misnamed actions on restore).
4. Issue Summary
| Number | Issue | Location | Severity |
|---|---|---|---|
| 1 | New scheduling/cycle-detection algorithms lack Javadoc on their method/class signatures | MultipleTableJobConfigParser.java (scheduleTransforms, ScheduledTransform, nextTransformActionIndex), ConfigParserUtil.java (checkTransformCycles, findTransformCycle), DryRunConnectValidator.java (scheduleTransforms, ScheduledTransform) |
Medium |
| 2 | --dry-run connect path doesn't call checkTransformCycles, so a cyclic config gets a generic "unresolved dependencies" message instead of the more specific "Transform dependency cycle detected: a -> b -> a" message the real submission path now gives |
DryRunConnectValidator.java (scheduleTransforms) |
Low |
5. Merge Recommendation
Conclusion: Ready to merge
- Blockers — none.
- Recommended fixes — non-blocking: Issue 1 (Javadoc on the new algorithms) and Issue 2 (align dry-run's cycle message with the real submission path) would both be reasonable small follow-ups, either in this PR or a fast-follow.
Overall: the core logic has now been independently re-verified byte-for-byte unchanged across three rounds of review (mine, twice, plus this one), backed by a worst-case test that actually proves the claimed legacy-compatibility property rather than just asserting "no exception thrown." CI (Build) was IN_PROGRESS on this exact head (2af00684c) at the time of this review — I'd ask for a final CI-green confirmation before a maintainer merges, but there is no outstanding code-side blocker from my side.
|
Thanks for pushing the update to
Once F1, F2/F4 and F6 have confirming tests and F7 has a docs note, I'm happy to approve. |
DanielLeens
left a comment
There was a problem hiding this comment.
Sorry for missing this in the last review — that's on me. @SEZ9's questions sent me back through scheduleTransforms line by line rather than answering from the review summary, and F2/F4 in particular turned up a real correctness gap my "Ready to merge" conclusion on 2af00684c missed. I'm re-opening the conclusion below; everything else in my prior review still stands.
Answering each point against the current head (2af00684c):
F1 (duplicate output IDs double-decrementing unresolvedInputCount) — No, there's no dedupe/guard. scheduleTransforms (MultipleTableJobConfigParser.java:578-648) keys waitingByInputId by output-id string, and the dependent-release loop at :641-647 is re-run once per transform that produces that id, not once per id. If two transforms both declare the same plugin_output, every dependent listed under that id gets unresolvedInputCount-- a second time when the second duplicate is scheduled. Because the "ready" check is == 0 (not <= 0), this doesn't cause a double-schedule of the same dependent, but it can make a dependent with multiple distinct missing inputs reach 0 (and get scheduled) after only one of its real inputs actually arrived — the other genuinely-missing input never gets produced in time, and parseTransform's own re-check (:499-509) throws missing scheduled inputs for it later. So it's real, just narrow: it needs two transforms sharing an output id and a third transform depending on both that id and something else. Low-to-medium likelihood in practice, but it's a genuine gap, not just theoretical — there's no validation anywhere that rejects duplicate plugin_output values before scheduling.
F2 / F4 (findLast fallback narrowed to single-transform jobs) — this is the one that changes my conclusion. The core problem isn't the singleTransformLegacyFallback/emptyInputFallback branch itself (MultipleTableJobConfigParser.java:658-671) — it's what actually reaches it. transform.inputIds is built via ConfigParserUtil.getInputIds (ConfigParserUtil.java:371-373):
static List<String> getInputIds(ReadonlyConfig config) {
return config.getOptional(PLUGIN_INPUT).orElse(Collections.singletonList(DEFAULT_ID));
}An omitted plugin_input never produces an empty list — it produces [DEFAULT_ID]. emptyInputFallback = transform.inputIds.isEmpty() (:662) is therefore only reachable when a config explicitly writes plugin_input: [], which is exactly what the one test exercising it, ConfigParserUtilTest#testRejectsAmbiguousEmptyInputFallbacks (:190-210), constructs. I checked every other test in ConfigParserUtilTest, MultipleTableJobConfigParserTest, and SeaTunnelConfValidateCommandTest — none of them omit plugin_input from a Config object; they all set it explicitly, including the ones with names like testPreservesSimpleGraphCompatibility (:107-118, single transform, still sets plugin_input=[legacy] explicitly). So the actual "user just leaves plugin_input out" path — the classic quick-start pattern used throughout the docs — has no test coverage at all in this PR, and it does not behave the way findLast() used to.
Trace a concrete case: source omits plugin_output → registered under DEFAULT_ID (MultipleTableJobConfigParser.java:396, same default). Transform 1 also omits plugin_input/plugin_output → its inputIds=[DEFAULT_ID] resolves against the source (fine), and because it also omits plugin_output, it re-registers under DEFAULT_ID too (:519-520, :575) — so far consistent with the old chaining behavior. Now add Transform 2 that explicitly sets plugin_output=stage1 (e.g. named for clarity) but the user's next transform, Transform 3, omits plugin_input expecting the old "bind to whatever came before" behavior. Transform 3's inputIds still defaults to [DEFAULT_ID] — but DEFAULT_ID was never overwritten by Transform 2 (which registered under stage1), so DEFAULT_ID still points at the original source entry. Transform 3 silently binds to the source, skipping Transform 2's output entirely, instead of failing loudly or chaining correctly. If nothing upstream ever produced DEFAULT_ID at all (e.g. the first transform in the chain used an explicit plugin_output), the failure mode flips to a hard Unable to resolve transform dependencies exception instead — because emptyInputFallback can't trigger (inputIds isn't empty, it's [DEFAULT_ID]) and singleTransformLegacyFallback requires transforms.size() == 1, which a multi-transform job never satisfies.
So this is a real regression for any multi-transform job that mixes an explicitly-named transform with a later transform relying on implicit chaining: either a silent wrong-binding (skips a stage) or a hard failure, depending on whether DEFAULT_ID happens to still be claimed by something. The legacy findLast() (queue-poll-based, dev HEAD prior to this PR) didn't have this issue because it always bound to the literal last-inserted map entry regardless of what key it was under — it never conflated "positionally last" with "the string DEFAULT_ID." I'd call this a High-severity blocker (Issue 3 below) — it's a data-correctness regression on a documented, commonly-used config pattern, not an edge case.
F3 (getTransformInputIds for scheduling vs. getInputIds for validation) — They don't currently diverge in behavior: DryRunConnectValidator.getTransformInputIds (:390-394) and ConfigParserUtil.getInputIds (:371-373) both do getOptional(PLUGIN_INPUT).orElse(singletonList(DEFAULT_ID)), byte-for-byte the same default. They're just duplicated implementations (same root cause as Issue 1/F5 below), so they can silently drift if one is edited without the other — worth unifying for that reason, not because they disagree today.
F5 (scheduler duplicated between DryRunConnectValidator and MultipleTableJobConfigParser) — Confirmed, and this is the same thing I flagged as Issue 1 in my last review: DryRunConnectValidator.scheduleTransforms (:227-307) and MultipleTableJobConfigParser.scheduleTransforms (:578-673) are independent near-duplicate implementations of the same Kahn's-algorithm scheduler (the engine copy adds the Fenwick-tree action-index bookkeeping the dry-run copy doesn't need). Neither calls the other. +1 to raising this — I'd already flagged it, just non-blocking on its own; it becomes more important now that F1/F2 show the duplicated logic has its own latent bugs, since a fix to one copy won't automatically fix the other.
F6 (single-transform self-cycle passing dry-run via the legacy fallback) — I checked this carefully because it sounds like a dry-run-vs-real divergence, but it isn't one: checkTransformCycles is only ever invoked from checkComplexGraph (ConfigParserUtil.java:81, via checkGraph:77-81), which only runs when !isSimpleGraph — and isSimpleGraph (:84-91) treats transforms.size() <= 1 as "simple" unconditionally. So a genuine single-transform self-reference (plugin_input pointing at its own plugin_output) never reaches checkTransformCycles on the real submission path either — it's a simple graph by definition, cycle-check is skipped, and it would hit the same singleTransformLegacyFallback path (transforms.size()==1 && !anyInputAvailable is true for a self-reference) and get silently accepted via findLast/legacy fallback on both the engine and dry-run copies. So dry-run and the real path are actually consistent here — this is pre-existing platform behavior (the isSimpleGraph gate), not something this PR introduced or diverges on.
F7 (docs/changelog) — Still outstanding, same as Issue 2 in my last review (dry-run's cycle message not matching the real path's). Agreed this needs a docs/changelog note for the new fail-fast JobDefineCheckException/ConfigCheckException behavior in general, beyond just that one message-format gap.
F8 (duplicated static helpers; unused getOutputId()) — Confirmed dead code: MultipleTableJobConfigParser.ScheduledTransform.getOutputId() (:763-765) and getActionIndex() (:767-769) have no callers anywhere in MultipleTableJobConfigParser.java, and the only test callers of a same-named getOutputId() (ConfigParserUtilTest.java:145,147, SeaTunnelConfValidateCommandTest.java:331) are exercising the other copies in ConfigParserUtil/DryRunConnectValidator, not this one. Low priority as noted, but worth removing or wiring in.
4. Revised Issue Summary
| Number | Issue | Location | Severity | Raised by another reviewer |
|---|---|---|---|---|
| 1 | Duplicated Kahn's-algorithm scheduler between MultipleTableJobConfigParser and DryRunConnectValidator, no shared implementation |
MultipleTableJobConfigParser.java:578-673, DryRunConnectValidator.java:227-307 |
Medium | Yes (@SEZ9, F5) |
| 2 | --dry-run connect doesn't call checkTransformCycles, so cyclic configs get a generic message instead of the specific cycle path |
DryRunConnectValidator.java (scheduleTransforms) |
Low | Partially — related to @SEZ9's F7 |
| 3 (new) | Omitted plugin_input resolves to [DEFAULT_ID], not an empty list, so emptyInputFallback never fires for it; a multi-transform job mixing an explicitly-named transform with a later implicit one either silently binds to the wrong (stale/original) table or hard-fails, instead of chaining to the true last-produced table the way the old findLast() did |
ConfigParserUtil.java:371-373, MultipleTableJobConfigParser.java:598,658-671 |
High | Yes (@SEZ9, F2/F4) — this reply is the first evidenced writeup |
| 4 (new) | Duplicate plugin_output values across two transforms cause waitingByInputId dependents to be decremented once per duplicate producer instead of once per id, which can release a dependent before all of its genuinely-distinct inputs are available |
MultipleTableJobConfigParser.java:589-610,641-647 |
Medium | Yes (@SEZ9, F1) |
5. Merge Recommendation
Conclusion: Ready to merge after fixes
-
Blockers — must be fixed
- Issue 3: the implicit-chaining regression. This needs either (a) distinguishing "genuinely absent
plugin_input" from "resolved to theDEFAULT_IDsentinel" so the fallback path can still track true positional last-output the wayfindLast()did, or (b) accepting the behavior change explicitly and documenting it as a breaking change with migration guidance (requireplugin_input/plugin_outputonce any transform in the job uses them). Given this is meant to preserve legacy compatibility (that's the PR's own stated goal), I'd lean towards (a).
- Issue 3: the implicit-chaining regression. This needs either (a) distinguishing "genuinely absent
-
Recommended fixes — non-blocking
- Issue 1/4 (duplication, duplicate-output double-decrement): worth fixing together since a fix to the scheduler for Issue 4 will need to land in both copies anyway.
- Issue 2: align the dry-run cycle message with the real path's.
Thanks again @SEZ9 for pushing on this instead of taking the summary at face value — the F2/F4 trace above wouldn't have surfaced from a diff read alone, it needed walking through what getInputIds's default actually resolves to. @goutamadwant, happy to help think through option (a) above if useful; the Fenwick-tree action-index machinery you already have for the non-fallback path shouldn't need to change, this is scoped to the fallback-detection condition itself.
Signed-off-by: goutamadwant <workwithgoutam@gmail.com>
|
Following up on F1 — duplicate output IDs double-decrementing
F2 / F4 — narrowed
F3, F5, F6, F7 and F8 from the earlier review are unchanged and still stand. Once F1 and F2/F4 are addressed with tests, I'm happy to re-review. |
|
Pushed d796232 after syncing dev. Runtime and dry-run now share dependency scheduling, repeated output IDs cannot release joins early, and supported omitted-input/default binding is preserved. Added regressions checking actual graph edges, schemas, legacy action indices and cycles. All 83 focused tests and scoped module verification passed on both Java 8 and 11. New CI is pending; the added container E2E has not been run locally. |
DanielLeens
left a comment
There was a problem hiding this comment.
Re-review from scratch on the current head (d796232d2954), 1 new commit since my last CHANGES_REQUESTED (2af00684c). I re-traced the algorithm end to end rather than trusting the PR's own summary of what changed.
What Problem Does This PR Solve?
The legacy transform-dependency resolver (MultipleTableJobConfigParser#parseTransforms, and its near-duplicate in DryRunConnectValidator) used a Queue<Config> retry loop: poll a transform, and if its plugin_input isn't ready, push it back to the tail and try the next one. Two real defects followed: (1) a transform whose input can never resolve (typo, forward reference, real cycle) spins the queue forever, hanging job submission / --dry-run connect; (2) a multi-input transform was treated "ready" the moment any one input resolved, silently dropping the rest instead of waiting, producing a DAG with missing edges.
The fix replaces the retry queue with an indexed, Kahn-style ready-scheduler that resolves a transform only once every declared input is available, adds explicit cycle detection, and reproduces the legacy retry-queue's exact poll-count via a Fenwick tree so default action names/indexes are byte-identical for every previously-valid graph (checkpoint/savepoint action-name compatibility across an upgrade).
1. Code Change Review
1.1 Core Logic Analysis — verifying my own CHANGES_REQUESTED blocker against the new commit
My last review flipped to CHANGES_REQUESTED on three real gaps surfaced by @SEZ9's F1/F2/F4 (I had missed them across three earlier passes). This commit (d796232d2954, "Share transform scheduling and preserve legacy input binding") is a targeted rewrite of the scheduler itself, not a superficial patch. I re-traced all three against the new code:
F1 — duplicate plugin_output double-decrementing unresolvedInputCount — fixed. The new shared TransformDependencyScheduler.scheduleTransforms (TransformDependencyScheduler.java:80-114) changed the release step from waitingByInputId.getOrDefault(...) (old code, read-only) to waitingByInputId.remove(transform.outputId) (new code). Because remove empties the map entry the first time any producer of that id is scheduled, a second transform producing the same plugin_output finds nothing left to release — dependents is null — so a dependent's unresolvedInputCount is decremented exactly once per distinct id, never once per producer. The comment at :110 states the invariant directly: "Each ID satisfies a dependency once, even when later producers overwrite its value." I hand-verified this is sufficient to prevent the exact narrow trigger I described previously (a dependent with two distinct missing inputs releasing early after only one duplicate-output producer runs).
- Regression test at the scheduler level:
ConfigParserUtilTest#testDuplicateOutputsReleaseEachMissingInputOnlyOnce— three inputs (dup×2,later) feeding ajoinedtransform; assertsjoinedis only scheduled (index 3) afterlater(index 2), not right after the firstdup. - Regression test at the full engine-wiring level (real
Actiongraph, not just scheduling order):MultipleTableJobConfigParserTest#testDuplicateUnnamedOutputsWaitForOtherJoinInputassertsjoined.getUpstream()actually contains both theDEFAULT_IDaction and thelateraction — i.e., the produced DAG edge is correct, not just that scheduling didn't throw. - Mirrored on the dry-run path:
SeaTunnelConfValidateCommandTest#testConnectDryRunDuplicateUnnamedOutputsWaitForOtherInputasserts all 6PluginResults areVALIDATEDand thatDryRunTestTransformFactory.getCreatedTables()showslaterwas created (and consumed) beforejoined.
This is exactly the "three-transform shape" @SEZ9 asked for, with tests at all three levels (isolated scheduler, real DAG wiring, dry-run wiring).
F2 / F4 — narrowed findLast/legacy fallback (the one that flipped my conclusion) — fixed. Root cause was that ConfigParserUtil.getInputIds resolves an omitted plugin_input to [DEFAULT_ID], never to an empty list, so the old emptyInputFallback = transform.inputIds.isEmpty() check could never fire for the common "just leave plugin_input off" quick-start pattern. The fix adds a new inputOmitted flag to ScheduledTransform (TransformDependencyScheduler.java:222, set from !readonlyConfig.getOptional(PLUGIN_INPUT).isPresent()) and changes the fallback condition to emptyInputFallback = transform.inputIds.isEmpty() || transform.inputOmitted (:130). This restores the "omitted input, nothing left to bind to, use the legacy last-inserted-table fallback" path without conflating it with a config that explicitly wrote plugin_input=[].
- I traced the concrete regression scenario from my last review by hand against the new code: source
src, transform 1 (plugin_input=[src], plugin_output=named), transform 2 (omitsplugin_input,plugin_output=terminal). Transform 2'sinputIds=[DEFAULT_ID]never resolves (nothing producesDEFAULT_ID), it's the sole unresolved transform at drain time,inputOmitted=true→emptyInputFallback=true→ legacy fallback fires, binding to the true last-inserted table (named's action), not silently to a stale/wrong entry. - This is backed by an unusually strong end-to-end test, not just an exception-absence check:
MultipleTableJobConfigParserTest#testTerminalOmittedInputUsesNamedTransformSchemaAndEdgebuilds real SQL transforms and assertsterminal.getUpstream()equalsnamed's actualAction, and that the resultingCatalogTable's row type has the column (name_copy) that only exists if the schema actually flowed fromnamed's output — i.e., it proves the correct edge and the correct schema, not just "didn't throw." Its siblingtestOmittedInputPrefersExistingDefaultOverLastNamedOutputcovers the complementary case whereDEFAULT_IDis actually available (e.g. an earlier transform used it) — correctly showing the omitted-input transform binds to that liveDEFAULT_IDentry in preference to the positionally-last named output, which is the historically-correct legacy semantics (default-id lookup first, "last inserted" only as a last resort). - Dry-run mirror:
SeaTunnelConfValidateCommandTest#testConnectDryRunOmittedInputUsesNamedSchema/testConnectDryRunOmittedInputPrefersExistingDefaultassertVALIDATEDstatus end to end for both branches. - The
docs/en(zh)/introduction/concepts/incompatible-changes.mdentry was rewritten in this same commit to describe the corrected semantics precisely: "An omittedplugin_inputstill resolves the default output ID first; it does not always select the preceding transform. When that ID is unavailable and only the omitted-input transform remains, the parser retains the legacy fallback to the last inserted table." I checked this against the code above and it's accurate, not aspirational.
F6 (from my last round, re-verified against the new commit) — single-transform self-cycle now genuinely rejected, on both paths. This wasn't part of my last CHANGES_REQUESTED, but the fix incidentally closes it too, so I re-checked it didn't regress. ConfigParserUtil.checkGraph's isSimpleGraph branch (ConfigParserUtil.java:73-76, new) now calls TransformDependencyScheduler.scheduleTransforms(transforms, singleton(sourceId)) before checkSimpleGraph, so a genuine 0-or-1-transform graph now goes through the same cycle-detection code as complex graphs. Inside the scheduler's cycle check, an explicit self-reference (plugin_input=[self], plugin_output=self, i.e. inputOmitted=false) registers a real self-edge and is rejected (TransformDependencyScheduler.java:270-274: the inputOmitted && inputId.equals(outputId) guard only suppresses the implicit omitted-chain case, not an explicit one). I confirmed this distinction is correct by tracing both branches:
- Explicit self-reference (
plugin_input=[self], plugin_output=self):inputOmitted=false→ real self-edge registered → Kahn's algorithm can't resolve it →JobDefineCheckException: Transform dependency cycle detected: self -> self. Tests:ConfigParserUtilTest#testSimpleGraphRejectsExplicitSelfCycle/#testSingleExplicitSelfCycleCannotUseLegacyFallback,SeaTunnelConfValidateCommandTest#testConnectDryRunRejectsSingleExplicitSelfCycle, and a genuine client-to-master E2E testJobExecutionIT#testRejectsExplicitTransformSelfCycleBeforeSubmissionthat submits through the realSeaTunnelClient/ClientJobExecutionEnvironmentand asserts the exception is thrown within a 30sassertTimeoutPreemptively— this is the single strongest piece of evidence against the original hang defect: this exact shape (a self-referencing transform) is precisely the kind of config that hung indefinitely under the old retry queue, and it's now asserted to fail fast through the real submission path, not just a unit-level scheduler call. - Implicit single-transform omission (classic quick-start
transform { sql { ... } }with noplugin_input/plugin_outputat all):inputOmitted=true, the self-edge guard suppresses it,checkTransformCyclessees zero transform-to-transform edges, no cycle, falls through to the legacy fallback path exactly as before. Test:ConfigParserUtilTest#testImplicitSingleTransformIsNotAnExplicitSelfCycleconfirmslegacyFallback=true,actionIndex=0.
This is a real, disclosed, and correctly-scoped tightening (previouslyisSimpleGraphskipped cycle detection unconditionally, so an explicit self-loop silently succeeded viafindLast) — and the doc entry now states it plainly: "An explicit self-reference with no previously available input is rejected too, including single-transform jobs."
1.2 Duplication (F5) — fixed, not just reduced
The entire scheduleTransforms/ScheduledTransform/Fenwick-tree/checkTransformCycles/findTransformCycle implementation (369 lines) has been extracted into a new shared seatunnel-engine-core class, TransformDependencyScheduler (public, in org.apache.seatunnel.engine.core.parse). MultipleTableJobConfigParser.parseTransforms now calls TransformDependencyScheduler.scheduleTransforms(...) directly and its own copies of the scheduler/Fenwick-tree code (208 lines) are deleted outright (verified via diff against 2af00684c, not just diffstat). DryRunConnectValidator.scheduleTransforms is now a 4-line wrapper that delegates to the same shared method and translates JobDefineCheckException to ConfigCheckException; its own independent ScheduledTransform, getTransformInputIds, getTransformOutputId, and unresolvedTransformDependencies are all deleted (confirmed via diff, not present in current head). This also closes F3 (divergent input-id resolvers) as a side effect — there is now exactly one getInputIds implementation (ConfigParserUtil.getInputIds, now public so TransformDependencyScheduler can call it), used by both the runtime and dry-run paths, so they cannot silently drift. This is the strongest structural improvement in this round: three separate reviewer-raised concerns (F2/F4 root cause, F5 duplication, F3 divergence risk) collapsed into one shared, single-sourced implementation rather than three independent patches.
1.3 Runtime Path (unchanged, re-verified against current head)
seatunnel.sh submit / --dry-run connect
-> MultipleTableJobConfigParser.parse() / DryRunConnectValidator.validate()
-> ConfigParserUtil.checkGraph(...) [isSimpleGraph -> NEW: scheduleTransforms + checkSimpleGraph;
complex -> checkComplexGraph -> scheduleTransforms]
-> parseTransforms(transformConfigs, ...)
-> TransformDependencyScheduler.scheduleTransforms(...) [shared Kahn's-algorithm ready queue + cycle check]
-> parseTransform(actionIndex, config, legacyFallback, ...) for each, in resolved order
Both the simple-graph and complex-graph branches, and both the real submission and dry-run entry points, now converge on the identical shared scheduler/cycle-check implementation. This is on the normal submission path (including savepoint restore, verified in earlier rounds and unchanged by this commit) for every job shape, not a boundary-only fix.
1.4 Compatibility Impact
Fully compatible for every graph that previously resolved successfully — action indexes/names are still preserved byte-for-byte, now additionally cross-checked by two new property-based tests: ConfigParserUtilTest#testFallbackAndDefaultBindingMatchLegacyQueuePolls (three hand-picked fixtures covering explicit-empty-input, omitted-input, and mixed cases) and #testShuffledValidChainsKeepLegacyQueuePollIndexes (100 randomized 32-transform reverse/scrambled dependency chains, each independently verified against a literal re-implementation of the legacy retry-queue in the test itself via assertLegacyQueueSchedule). This is meaningfully stronger than the single hand-traced 512-node fixture I verified in earlier rounds — it's now a fuzz-style check across 100 random valid topologies.
The genuinely incompatible behavior (multi-transform unresolvable input now fails fast, cycles now rejected including single-transform self-references) remains correctly and specifically documented in both docs/en and docs/zh incompatible-changes.md, updated in this same commit to match the corrected fallback semantics — I checked the current text against the code trace above and it is accurate, including the subtlety that an omitted plugin_input "does not always select the preceding transform" (it prefers a live DEFAULT_ID entry first).
1.5 Performance / Side-Effect Analysis
Unchanged from previous rounds: single-threaded, config-parse-time-only code path (client/master submission or dry-run, not per-record), O((V+E) log V) scheduling, no new locks/threads/unbounded structures. The waitingByInputId.remove() change (vs. the old getOrDefault, read-only) is O(1) amortized and strictly reduces work (one less map entry retained after release) — no new performance risk from the F1 fix.
1.6 Error Handling and Logging
Unchanged and correct: exceptions carry the concrete unresolved id(s), missing input ids, and available output ids; overflow is explicitly guarded; no exceptions swallowed; no sensitive data logged.
2. Code Quality Assessment
2.1 Coding Standards
TransformDependencyScheduler carries a class-level Javadoc explaining its contract in one dense paragraph ("Resolves every distinct input before plugin discovery in O((V + E) log V) time. Circular selection and Fenwick poll counts preserve the legacy queue's action indexes...") — this addresses the Medium-severity Javadoc gap (Issue 1) I raised two rounds ago, at the class level; the individual methods (scheduleTransforms, nextTransformActionIndex, the Fenwick helpers, checkTransformCycles, findTransformCycle) still rely on inline // comments rather than per-method Javadoc. Non-blocking — the class-level doc plus the inline comments (which are genuinely explanatory, not restating code) are sufficient for a reader to reconstruct the algorithm, but a per-method Javadoc would still be a reasonable low-cost follow-up.
ScheduledTransform's fields remain uncommented individually, though the class is now public static final (necessarily, to be shared across modules) with a one-line class Javadoc ("A transform in evaluation order, with its legacy poll index and input-binding mode.") — adequate given the narrow, well-tested usage.
2.2 Test Coverage and Test Stability
Substantially expanded this round, and — importantly — expanded in the direction that actually matters: several new tests assert real Action.getUpstream() edges and real CatalogTable schemas (MultipleTableJobConfigParserTest), real PluginResult.VALIDATED status end-to-end (SeaTunnelConfValidateCommandTest), and a real client-to-master submission with a wall-clock timeout guard (JobExecutionIT) — not just "the scheduler returns some list without throwing." This is exactly the kind of coverage that would have caught F1/F2/F4 had it existed before the previous CHANGES_REQUESTED round, and it now directly targets the same three defects. All new tests are deterministic: in-memory Config/TreeSet/LinkedHashMap iteration for the unit-level tests, a seeded Random(12079L) for the property-based fuzz test (reproducible), and a bounded assertTimeoutPreemptively(Duration.ofSeconds(30), ...) for the one container-free E2E test rather than an unbounded wait. No Thread.sleep, no flaky timing assumptions.
Rating: Stable.
2.3 Documentation Updates
Both docs/en and docs/zh incompatible-changes.md entries were rewritten in this commit to describe the corrected fallback/cycle semantics precisely (exception types named explicitly, self-reference-in-single-transform-jobs called out, the "default ID first, last-inserted-table only as fallback" nuance spelled out). I cross-checked the English text against the code trace above and it holds. I did not do a line-by-line bilingual diff against the Chinese version this round (I did in an earlier round and it was a faithful translation); given the English content changed substantively here, I'd ask the author or another reviewer to confirm the docs/zh wording was updated in lockstep rather than assume it from my earlier check — this is Low severity since the English (authoritative) text is correct and both files were touched in the same commit per the diffstat.
3. Architectural Soundness
3.1 Elegance of the Solution
The consolidation into TransformDependencyScheduler is a materially better architecture than what I approved in my "Ready to merge" rounds — it doesn't just fix the two narrow bugs F1/F2/F4 exposed, it removes the structural precondition that let them exist in the first place (two independently-maintained copies of a subtle algorithm, with two independently-maintained default-id resolvers). This is root-cause remediation of the review feedback, not a patch bolted onto the flagged lines.
3.2 Maintainability
Significantly improved. The single-source TransformDependencyScheduler means a future fix only needs to land once, and the new fuzz-style test (testShuffledValidChainsKeepLegacyQueuePollIndexes) gives meaningfully more confidence than a single hand-picked adversarial fixture that a future change to the scheduler won't quietly break legacy-index parity.
3.3 Extensibility
Unchanged assessment: ScheduledTransform cleanly separates configured input from resolved position; would extend cleanly for richer per-edge metadata later.
3.4 Historical-Version Compatibility
This remains the property the PR is most careful about, and it's now backed by both the original 512-node adversarial test and a new 100-iteration randomized property test. ActionStateKey-based checkpoint/savepoint restore across an upgrade depends on action.getName() staying stable for every valid pre-existing DAG, and that property continues to hold and is now more rigorously tested than in any previous round.
4. Issue Summary
| # | Issue | Location | Severity | Raised by another reviewer |
|---|---|---|---|---|
| 1 (was F1) | (Resolved in d796232d2954) Duplicate plugin_output values double-decremented unresolvedInputCount once per producer instead of once per id. Fixed via waitingByInputId.remove() (release-once semantics) plus regression tests at scheduler, engine-wiring, and dry-run levels. |
TransformDependencyScheduler.java:107-114; tests in ConfigParserUtilTest, MultipleTableJobConfigParserTest, SeaTunnelConfValidateCommandTest |
Resolved (was High/Medium — narrow trigger, real gap) | Yes (@SEZ9, F1) |
| 2 (was F2/F4) | (Resolved in d796232d2954, this is the blocker that drove my last CHANGES_REQUESTED) Omitted plugin_input resolved to [DEFAULT_ID], never an empty list, so the legacy terminal fallback never fired for the common "just omit plugin_input" pattern in a multi-transform job — causing silent wrong-binding or a hard failure. Fixed via the new inputOmitted flag folded into emptyInputFallback, restoring "prefer live DEFAULT_ID, else fall back to the last-inserted table" — the actual legacy semantics, not just "don't throw." Verified end to end with real Action.getUpstream()/schema assertions on both the runtime and dry-run paths. |
TransformDependencyScheduler.java:130,222; tests in ConfigParserUtilTest, MultipleTableJobConfigParserTest, SeaTunnelConfValidateCommandTest |
Resolved (was High) | Yes (@SEZ9, F2/F4) — my own writeup two rounds ago |
| 3 (was F5) | (Resolved) Scheduler logic duplicated between MultipleTableJobConfigParser and DryRunConnectValidator. Fully extracted into shared TransformDependencyScheduler; both call sites delegate to the identical implementation. |
TransformDependencyScheduler.java (new), MultipleTableJobConfigParser.java, DryRunConnectValidator.java |
Resolved | Yes (@SEZ9, F5; also my Issue 1/2 from two rounds prior) |
| 4 (was F3) | (Resolved as a side effect of Issue 3) getTransformInputIds (dry-run) vs. getInputIds (runtime) were independent implementations that could silently drift. Now a single ConfigParserUtil.getInputIds (made public) backs both paths. |
ConfigParserUtil.java:266 |
Resolved | Yes (@SEZ9, F3) |
| 5 (was F6) | (Resolved, not part of my last CHANGES_REQUESTED but re-verified this round) isSimpleGraph previously skipped cycle detection unconditionally, so an explicit single-transform self-reference silently succeeded via the legacy fallback. Now checkGraph's simple-graph branch also runs the shared scheduler/cycle-check first; an explicit self-reference is rejected on both the real submission path (including a genuine client-to-master E2E test with a 30s timeout guard) and dry-run, while the classic implicit single-transform omission pattern is correctly preserved as non-cyclic. |
ConfigParserUtil.java:73-76, TransformDependencyScheduler.java:270-274 |
Resolved | Yes (@SEZ9, F6) |
| 6 (was F7) | (Resolved) Docs/changelog needed to reflect the fail-fast exception behavior. incompatible-changes.md (en/zh) rewritten in this same commit with accurate, specific language including exception types and the self-reference/omitted-input nuances. |
docs/en(zh)/introduction/concepts/incompatible-changes.md |
Resolved | Yes (@SEZ9, F7) |
| 7 (was F8) | (Effectively resolved) ScheduledTransform.getOutputId()/getActionIndex() had no production callers. They're now exercised extensively by the new test suite across all three test files; still no production caller of getOutputId() specifically, but this is no longer meaningfully "dead code" and isn't worth blocking on. |
TransformDependencyScheduler.java (ScheduledTransform) |
Low (non-blocking) | Yes (@SEZ9, F8) |
| 8 (new, minor) | No production code rejects duplicate plugin_output values outright; the F1 fix makes duplicates safe (release-once semantics) but a downstream consumer of a duplicate id still binds to whichever producer's entry is last in tableWithActionMap at wiring time (map-overwrite semantics), which is ambiguous-but-harmless pre-existing behavior, not something this commit changes or worsens. Worth a follow-up validation (reject duplicate plugin_output with a clear error) but out of scope for this fix and not a regression. |
MultipleTableJobConfigParser.java (tableWithActionMap.put, pre-existing) |
Low (non-blocking, pre-existing, not a regression) | No |
| 9 (new, minor) | New algorithm methods (scheduleTransforms, Fenwick helpers, checkTransformCycles) still rely on inline comments rather than per-method Javadoc; the new class-level Javadoc on TransformDependencyScheduler substantially closes this gap but doesn't fully resolve my earlier per-method Javadoc ask. |
TransformDependencyScheduler.java |
Low (non-blocking) | Partially — carryover of my own earlier Issue 1 |
No High or Medium severity issues remain open. Every item that drove my CHANGES_REQUESTED two rounds ago (Issues 1–6 above) has confirming evidence in the current head, not just a claim in the commit message.
5. Merge Recommendation
Conclusion: Ready to merge after CI
- Blockers — must be fixed: None from a correctness/compatibility standpoint. My
CHANGES_REQUESTEDconclusion from2af00684cis resolved — I independently re-traced F1 and F2/F4 (the two gaps that changed my prior conclusion) against the newd796232d2954code rather than trusting the commit message, and both are fixed correctly with meaningfully strengthened regression coverage (real DAG-edge/schema assertions, not just exception-absence checks, plus a genuine client-to-master E2E test with a bounded timeout that reproduces the original hang shape and confirms it now fails fast). F3/F5/F6/F7/F8 from @SEZ9's list are also resolved, several as a structural side effect of extracting the sharedTransformDependencyScheduler. - CI status: the fork's
Buildworkflow ford796232d2954wasqueued(not yet run) at the time of this review, and the Apache-side status check showspending. I have not seen a completed run for this exact commit, so I cannot independently confirm compile/test success — I'm relying on the author's own report of "83 focused tests ... passed on both Java 8 and 11" plus my own line-by-line trace of the diff and the new tests' assertions, not a green CI badge. A maintainer should wait for this run to complete before merging; if it comes back red, that supersedes this review. - Recommended fixes — non-blocking: Issue 7 (Low, non-blocking
getOutputId()still unused in production), Issue 8 (Low, consider rejecting duplicateplugin_outputvalues explicitly in a follow-up), Issue 9 (Low, per-method Javadoc would still help).
Overall assessment: this is a genuine, verified resolution of the correctness gap that justifiably reopened my earlier "ready to merge" conclusion. Rather than patching around the two narrow bugs F1/F2/F4 exposed, the author addressed their structural root cause (duplicated, independently-drifting scheduler implementations) by extracting a single shared TransformDependencyScheduler, and backed every fix with tests that assert real DAG wiring and schemas, not just absence of an exception — including a new client-to-master E2E test that reproduces the original hang shape under a wall-clock timeout. I found no new correctness gap in my independent trace of this commit. The only remaining open item is CI completion, which I could not confirm at review time.
Signed-off-by: goutamadwant <workwithgoutam@gmail.com>
|
Synced dev's merged CI fixes; all 83 focused graph/parser/CLI tests pass on Java 11. The graph patch is unchanged. Savepoint expiry and checkpoint-scheduler cleanup remain unresolved investigations, not fixes claimed by this sync. Fresh CI is queued. |
|
Thanks for the sync update, @goutamadwant. Since the graph patch itself is unchanged, the previously raised points on
On the test claim: the 83 focused tests passing is good to hear, but none of the items above appear to be covered by them as far as I can tell from the thread — if some are, please point to the specific test names. Also, since savepoint expiry and checkpoint-scheduler cleanup are outside this PR's scope and still under investigation, please keep them out of this change so we can review the graph fix on its own. I'll take another look once the fresh CI run and the updates above land. |
|
Rechecked F1-F8 against the current head. F1's omitted-output case is already covered by testConnectDryRunDuplicateUnnamedOutputsWaitForOtherInput and testDuplicateUnnamedOutputsWaitForOtherJoinInput; the keyed waiting list is removed after its first release. F2/F4 cover terminal omitted-input fallback and default-first binding. F3/F5 use the same scheduler/input parser in runtime and dry-run. F6 rejects explicit self-cycles while preserving implicit legacy binding; F7 has EN/ZH compatibility notes. For F8, getOutputId still has test callers; I propose leaving that optional API cleanup separate. No additional scheduler rewrite is needed for these cases. This source/test mapping is not a claim that unrelated CI failures are resolved. |
|
Thanks for the F1–F8 recheck, @goutamadwant — the mapping is helpful. A few follow-ups so I can close these out:
Understood that this mapping doesn't claim anything about the unrelated CI failures. Once the items above are answered I'll do a final pass. |
Signed-off-by: goutamadwant <workwithgoutam@gmail.com>
…ycle Signed-off-by: goutamadwant <workwithgoutam@gmail.com>
DanielLeens
left a comment
There was a problem hiding this comment.
Hi @goutamadwant, thanks for the latest sync with dev. The head moved to ad8cadd12, so per our process I re-read the full diff on the new head rather than only the merge delta. Short version: the PR's own diff is unchanged, my earlier conclusion stands, and I have no new source-side concern.
What Problem Does This PR Solve?
- User pain point: a cyclic transform graph (including an explicit self-reference) could hang job submission or
--dry-run connect, and a multi-input transform could be created from only the inputs that happened to be available, silently dropping the remaining upstreams. - Fix approach: one shared
TransformDependencySchedulerused by the runtime parser, the graph pre-check and the dry-run validator, with fail-fast errors for unresolved or cyclic graphs. - One-sentence summary: transform dependency resolution becomes one deterministic, fail-fast algorithm instead of two retry queues that could hang or mis-bind inputs.
1. Code Change Review
1.1 Core Logic Analysis
- What changed since my last review (
7bb568104). Only a merge ofdev(merge commitad8cadd12plus thedevcommits it brings in). I diffed all 12 non-doc PR-owned files between7bb568104andad8cadd12and the diff is empty. The only PR-owned files that differ are the twoincompatible-changes.mdfiles, and those differences are just other entries that landed ondev(the Java 11 runtime note and the Doris timestamp-timezone note); the "Transform Dependency Resolution" entry is still present in both EN and ZH. The PR's diff against currentdevis still the same 14 files (+1598/-81) and the branch is 0 commits behinddevwith no conflicts. - Runtime path (unchanged, re-verified).
MultipleTableJobConfigParser.parse -> ConfigParserUtil.checkGraph -> TransformDependencyScheduler.scheduleTransforms, thenparseTransformper scheduled transform (plugin discovery only after inputs exist). The dry-run pathDryRunConnectValidator.validateTransforms -> scheduleTransformscalls the same scheduler and translatesJobDefineCheckExceptionintoConfigCheckException. Duplicate default outputs release a waiting join exactly once, the omitted-input fallback stays distinct from an explicitplugin_input, and an explicit self-reference ends in a concrete cycle error instead of looping.
1.2 Compatibility Impact
Partially incompatible, intentionally and documented. Valid graphs keep the same evaluation order and action indexes. Graphs that previously hung or silently mis-bound now fail fast at submission. The behavior change is documented in docs/en and docs/zh incompatible-changes.md with migration guidance. No config option, default or serialization format changes.
1.3 Performance / Side-Effect Analysis
Scheduling runs once per submission or dry-run in O((V + E) log V) and never touches the record path. This removes an unbounded-retry CPU risk rather than adding one.
1.4 Error Handling and Logging
Errors carry the unresolved output IDs, the available IDs and a plugin_input / plugin_output hint, or the concrete cycle path. Nothing is swallowed. No new issues.
2. Code Quality Assessment
- Coding standards: the non-obvious parts of the scheduler are documented.
- Test coverage and stability:
ConfigParserUtilTest,MultipleTableJobConfigParserTest,SeaTunnelConfValidateCommandTestandJobExecutionIT#testRejectsExplicitTransformSelfCycleBeforeSubmission(bounded byassertTimeoutPreemptively) are deterministic and unchanged. Stability rating: Stable. - Documentation: EN and ZH are both updated and consistent.
3. Architectural Soundness
Long-term solution: one scheduler and three call sites, with no duplicated retry logic.
4. Issue Summary
No blocking issues found. No new issues from this pass.
5. Merge Recommendation
Conclusion: Ready to merge
- Blockers: none on the source side.
- Recommended fixes: none.
Existing threads. The open @SEZ9 request (assert the self -> self message in the dry-run self-cycle test) is already satisfied in SeaTunnelConfValidateCommandTest (testConnectDryRunRejectsSingleExplicitSelfCycle), and the F1 to F8 evidence in the earlier comments still applies line for line because the code is unchanged.
CI. On this head the Build check was still queued when I looked, so I cannot call it green yet. On the previous head (7bb568104) the red jobs were unrelated to this change (ElasticsearchIT slice tests that also fail on dev, fix open as #12711; the SplitClusterFaultToleranceIT worker-loss-during-cancel race, engine fix open as #12311; and an Azure Cosmos DB emulator connection failure). Neither fix is on dev yet, so please do not patch unrelated tests in this PR; if the new run shows the same failures, re-running only the failed jobs is the lowest-cost step. The remaining gates are CI and a write-capable maintainer.
Thanks again for the careful, patient work on this one.
|
Thanks for syncing with On this head the earlier points still hold as resolved: the runtime parser, the graph pre-check and the dry-run validator share one My earlier conclusion stands and I have no new source-side concern from this sync. |
|
Quick CI update for @goutamadwant. In my last review (head The failed jobs:
Suggested next step: merge the latest My source-side conclusion is unchanged: no source blockers, ready to merge once CI is green and a write-capable committer approves. |
|
Thanks for the CI breakdown on The suggested path sounds right to me: merge the latest Once CI is green after the |
DanielLeens
left a comment
There was a problem hiding this comment.
Re-reviewed at current head 612b5ede51 (verified immediately before posting).
I traced both runtime parsing and connect dry-run through the shared TransformDependencyScheduler. It records every distinct missing input before plugin discovery, releases each waiting set only once when its output becomes available, and preserves the legacy retry queue ordering/action indices through the Fenwick count. It also keeps the two deliberate legacy fallbacks narrowly scoped: a final omitted/explicit-empty input and the single-transform mismatch. The remaining graph now fails deterministically with either the concrete cycle path or unresolved input IDs; it can no longer construct a multi-input transform from a partial input set.
The core and dry-run paths are covered for cycles, unresolved/mixed inputs, reverse declaration, duplicate IDs, explicit-empty/single-transform compatibility, and index boundaries. I found no source-level blocker.
Merge gate: required Build is currently FAILURE. Please resolve or rerun it successfully for this exact head before merging.
Conclusion: code is ready for maintainer approval after a green Build. COMMENT review only because my reviewer account is read-only.
|
Thanks for the follow-up work on this. Based on the re-review at Looks addressed (pending my own pass over the head):
Still open — please reply or push:
Once those are in, I'll do a last read of the head and approve. |
|
Thanks @SEZ9 for the careful follow-up. I re-checked each open item against the current head 1. F3 (single resolution path). Transforms resolve their IDs exactly once, in the 2. F7 (docs). Present in both languages: 3. F8. 4. Build. The required On my side there is no remaining source-side blocker, and I have not run anything locally. From here it only needs your final read and a write-capable committer to merge. Thanks again to @goutamadwant for the patient iteration on this one. |
DanielLeens
left a comment
There was a problem hiding this comment.
Thank you for syncing dev again, @goutamadwant. This is a full re-review of the entire current diff at head b190936b and of the runtime call chain, not only of the merge. The PR-owned source and test files are byte-identical to the previous head 612b5ede; the only PR files that differ are the two incompatible-changes.md files, which picked up unrelated entries from dev (checked with a compare between the two heads).
What Problem Does This PR Solve?
- User pain point: In Zeta job parsing and
--dry-run connect, a cyclic transform graph made the retry queue spin forever, and a multi-input transform was built as soon as any one of its configured inputs existed, silently dropping the other upstream edges (#12077). - Fix approach: One shared
TransformDependencySchedulerreplaces both retry queues. It indexes waiting transforms per missing input, releases each set exactly once, preserves the legacy evaluation order and action indexes with a Fenwick tree, detects cycles with Kahn's algorithm, and delays plugin discovery until all inputs exist. - One-sentence summary: Cycles and unresolved inputs now fail fast with a concrete message, multi-input transforms always keep all configured upstreams, and valid graphs keep their order, names and indexes.
1. Code Change Review
1.1 Core Logic Analysis
Before (MultipleTableJobConfigParser.parseTransform, simplified):
inputs = inputIds.stream().map(tableWithActionMap::get).filter(Objects::nonNull)...;
if (inputs.isEmpty()) { if (transforms.isEmpty()) { inputs = findLast(...); } else { transforms.offer(config); return; } }After (MultipleTableJobConfigParser.java:468-524, TransformDependencyScheduler.java:53-151):
List<ScheduledTransform> scheduled = scheduleTransforms(transformConfigs, tableWithActionMap.keySet());
// each transform is parsed only after every distinct plugin_input exists; plugin jars are resolved afterwards (:520)Runtime path:
Client: ClientJobExecutionEnvironment.getLogicalDag -> MultipleTableJobConfigParser.parse
-> ConfigParserUtil.checkGraph (simple graph :72-78 / complex graph phase 6 :175-180)
-> parseSource -> parseTransforms -> TransformDependencyScheduler.scheduleTransforms (:471) -> parseTransform per scheduled item
CLI: DryRunConnectValidator.validateTransforms (:215) -> same scheduler -> validateTransform
Key findings (verified on this head):
TransformDependencyScheduler.java:116-124: dependents wait inwaitingByInputIdand are released throughremove(outputId), so each ID satisfies a dependency exactly once even when several transforms produce the same output ID (the earlier double-decrement concern is closed by this).:127-150: after the ready loop, remaining transforms go first throughcheckTransformCycles(explicit self-edges and real cycles throw, the omitted-input implicit chain is exempt at:292), then through the two documented fallbacks (a single remaining omitted/empty-input transform, or a one-transform job where no configured input exists). Everything else fails with the unresolved IDs listed.- The Fenwick bookkeeping (
:166-199) reproduces the old poll-derived action index, so default action names stayTransform[n]-factory. As an independent cross-check I modelled the retry queue (all-inputs readiness) and this scheduler's ready/queue-head/Fenwick logic in a small script and compared order and action index on 200,000 random DAGs, including reverse declaration order and duplicate output IDs: no difference, and cyclic cases were rejected by both. That is my own model of the algorithm, not the repository code, so I treat it as supporting evidence only. - Plugin discovery moved after readiness (
MultipleTableJobConfigParser.java:520), so a transform that is not ready no longer triggers repeatedgetTransformPluginJarPathslookups. ConfigParserUtil.java:72-78and:175-180run the scheduler during graph checking, so a cycle is rejected before any DAG is built on the client; the ID set passed is the set of source output IDs, which matchestableWithActionMapat the start ofparseTransforms.
1.2 Compatibility Impact
Partially incompatible, intentionally and documented (docs/en/introduction/concepts/incompatible-changes.md, "Transform Dependency Resolution", with the zh counterpart):
- Valid graphs where every
plugin_inputresolves keep their evaluation order, action indexes, default names and ordering ofinputs; duplicate input IDs behave as before. - Configs that relied on the old bugs now fail at submission: a multi-input transform with an input ID that never appears (typo, previously silently dropped), mutually dependent transforms (previously an endless loop), and several unresolved omitted-input transforms (previously an endless loop).
- The two legacy fallbacks are retained and narrowly scoped, so older single-transform configs with a mismatched ID, and a final omitted-input transform, still work.
- No option, API or wire change, and no persisted-state change: the logical DAG is produced at submission and serialized with the job, so running jobs and master failover are not re-parsed. A resubmission, including a restore that re-submits the same config, is parsed again (see issue 2).
1.3 Performance / Side-Effect Analysis
Parsing cost drops from a repeated retry loop (unbounded for cycles, quadratic polling for reverse-ordered chains, with plugin discovery per poll) to O((V + E) log V) time and O(V + E) memory, executed once per parse and once more in graph checking. V is the number of transforms in a job, so this is negligible at any realistic job size, and nothing here runs per record, so the production data volume does not matter for this change. No new threads, locks or retries; no secrets are logged (the error message lists output/input IDs and available IDs only).
1.4 Error Handling and Logging
Failures raise JobDefineCheckException on the Zeta path and are re-wrapped as ConfigCheckException in dry-run (DryRunConnectValidator.java:224-231), both with the cycle path or the unresolved output <- inputs pairs. Issues:
- Low - CI is red for reasons outside this diff, and it did not execute this PR's own unit tests on this head. The four
unit-testlegs fail inconnector-jdbcatJdbcOptionRuleCompletenessTest.testSinkRuleDeclaresAllSinkOptions(unknown optionnumeric_special_value_as_null), the knowndevbreak tracked in #12874 with the fix in #12875 (still open). Because the reactor stops there,Engine : Core,Engine : Client,Core : StarterandEngine : Servershow as SKIPPED in those jobs.jdbc-connectors-it-part-1 (17)failed while resolving theos-maven-pluginbuild extension (an artifact-resolution problem), and the twomysql-cdcjobs were cancelled. The jobs that cover this change did pass on this head:engine-v2-it(11 and 17, including the newJobExecutionIT#testRejectsExplicitTransformSelfCycleBeforeSubmission),transform-v2-it-part-1/2,engine-k8s-itandengine-resource-managers-it. The previous head had a greenunit-testrun, and the PR-owned files have not changed since. Best improvement: once #12875 is merged, syncdevand rerun only the failed jobs so the unit-test legs reach the engine and starter modules on the final head. Raised by another reviewer: no. - Low - Document the savepoint and resubmission implication (new on my side). Location: the "Transform Dependency Resolution" entry in
incompatible-changes.md. A job that is resubmitted from a savepoint with a config that contains an unresolved or partialplugin_inputwill now fail at parse until the config is corrected, and once corrected the previously dropped upstream feeds the transform, which changes the data a stateful transform sees. This is the intended direction, but a sentence in the migration guide would save operators a surprise. Sorry for not raising this in earlier rounds. Raised by another reviewer: no. - Low - Comment the fallback decision (new on my side). Location:
TransformDependencyScheduler.java:136-149. TheemptyInputFallback || singleTransformLegacyFallbackbranch is the single place where an otherwise ambiguous graph is accepted; a short comment saying whyanyInputAvailableis part of the single-transform rule (a one-transform job with some existing input is still treated as a real mismatch) would make the contract obvious to the next reader. Sorry for not raising this earlier. Raised by another reviewer: no.
2. Code Quality Assessment
2.1 Coding Standards
Good: the class and the main method carry Javadoc stating complexity and fallback scope, the Fenwick section is commented with its purpose, and the nested ScheduledTransform hides its state behind getters. ConfigParserUtil.getInputIds became public only so that the starter module reuses the same input resolution (no divergent re-parse remains, which closes the earlier single-resolution-path question); the only other getInputIds use in DryRunConnectValidator is on the sink path.
2.2 Test Coverage and Test Stability
Strong: ConfigParserUtilTest, MultipleTableJobConfigParserTest, SeaTunnelConfValidateCommandTest and a new JobExecutionIT case cover two-node and self cycles, partially and wholly unresolved inputs, mixed source/transform inputs, duplicate input IDs and duplicate unnamed outputs, valid reverse-declared chains, terminal omitted-input binding, explicit-empty-input fallback, action-index overflow, and both the runtime and the dry-run entry points. Hang regressions use assertTimeoutPreemptively, so a returning infinite loop turns the test red instead of hanging CI. The only deletions in test code are a relocated mockCatalogTable helper and a replaced OptionRule line in a test factory that gains an optional expected_table, so no assertion was weakened. Stability rating: Stable (no Thread.sleep; timeouts are generous at 30 seconds against millisecond work).
2.3 Documentation Updates
en and zh incompatible-changes.md are both updated with behavior, affected components and migration guidance. See issue 2 for the one missing sentence.
3. Architectural Soundness
3.1 Elegance
Long-term solution: one scheduler in engine-core shared by the runtime parser, the graph pre-check and the dry-run validator, instead of three divergent retry loops.
3.2 Maintainability
Much better than before; the Fenwick tree is the one non-obvious piece and it is documented and pinned by the index-boundary tests.
3.3 Extensibility
Adding a new fallback or dependency source means touching one class.
3.4 Historical-Version Compatibility
Only parsing at submission changes. No serialized job, checkpoint or savepoint format is touched, so there is no upgrade-path impact beyond the documented stricter validation of previously ambiguous configs.
4. Issue Summary
| Number | Issue | Location | Severity |
|---|---|---|---|
| 1 | Build is red from the pre-existing JDBC break; unit-test legs never reach engine-core, client and starter on this head | fork Build run, connector-jdbc JdbcOptionRuleCompletenessTest |
Low (merge gate) |
| 2 | Migration guide should mention savepoint resubmission with unresolved or partial inputs | docs/*/introduction/concepts/incompatible-changes.md |
Low |
| 3 | Comment why the single-transform fallback depends on anyInputAvailable |
TransformDependencyScheduler.java:136-149 |
Low |
5. Merge Recommendation
Conclusion: Ready to merge after fixes
Blockers:
- A green
Buildon the final head (syncdevafter #12875 lands and rerun the failed jobs). I have no source-level blocker.
Recommended fixes (non-blocking):
- Add the savepoint/resubmission sentence to the migration guide (issue 2).
- Add the short fallback comment (issue 3).
Overall assessment: A careful, well-tested fix for a real hang and a silent data-lineage bug, with the legacy order and names preserved for valid graphs. The change is exercised on every job submission, but its blast radius is limited to parse-time validation and it is covered by unit and end-to-end tests.
Better alternative: None that is simpler without reintroducing the retry loop; keeping one shared scheduler is the right shape.
|
Thanks for the detailed re-check against head The locations you listed for F3 (single resolution path in the No further questions from my side for now; I'll do the final read-through on |
|
Thanks @SEZ9. The earlier green run you cited is for My previous source-level conclusion remains clear, with the migration/comment notes non-blocking; your direct verification remains yours to close. Separately, GitHub currently reports real conflicts against dev. Please resolve those conflicts, and once the shared JDBC correction is available, include it and rerun the affected checks on the resulting head. I will re-review the full updated diff after a new commit. This follow-up does not repeat the earlier source review or declare current CI green. Current status evidence: head CI references: current Build check, Run / unit-test (11, ubuntu-latest), Run / jdbc-connectors-it-part-1 (17, ubuntu-latest). |
DanielLeens
left a comment
There was a problem hiding this comment.
Thanks @goutamadwant for keeping this branch in sync with dev again. The head moved to 3696a4b3c07d after my last review at b190936bbbcb, so this is a fresh full re-read of the PR rather than a look at the merge delta only. Analysis is static; nothing was built or run locally.
What Problem Does This PR Solve?
- User pain point: In Zeta job parsing and
--dry-run connect, a cyclic transform graph made the retry queue spin forever, and a multi-input transform was built as soon as any one of its configured inputs existed, silently dropping the other upstream edges (#12077). - Fix approach: one shared
TransformDependencySchedulerinengine-corereplaces the retry queues in the runtime parser, the graph pre-check and the dry-run validator. It releases each waiting set exactly once, keeps the legacy evaluation order and action indexes, detects cycles, and delays plugin discovery until every input exists. - One-sentence summary: cycles and unresolved inputs now fail fast with a concrete message, and valid graphs keep their order, names and indexes.
1. Code Change Review
1.1 Core Logic Analysis
What changed since b190936bbbcb: I diffed every PR-owned source, test and E2E file between the two heads and they are byte-identical. The only PR file that differs is docs/en/introduction/concepts/incompatible-changes.md, which gained other teams' entries from dev (Markdown child_ids, ARRAY_MAX/MIN, DataValidator LENGTH, IMap WAL). The "Transform Dependency Resolution" entry (incompatible-changes.md:179) is unchanged. So my earlier analysis of the code still holds on this head:
Client: MultipleTableJobConfigParser.parse
-> ConfigParserUtil.checkGraph (cycle / unresolved pre-check)
-> parseTransforms -> TransformDependencyScheduler.scheduleTransforms -> parseTransform (plugin discovery only after readiness)
CLI: DryRunConnectValidator.validateTransforms -> same scheduler -> validateTransform
- Each output ID releases its waiters once, so duplicate output IDs cannot release a multi-input transform early.
- Only after the ready loop does it check cycles, then apply the two documented fallbacks (
TransformDependencyScheduler.java:136-149), then fail with the unresolvedoutput <- inputspairs. - Dry-run and runtime share one resolution path, so they cannot disagree.
1.2 Compatibility Impact
Partially incompatible by design and documented in en and zh: configs that relied on the old hang or on silently dropped inputs now fail at submission. Valid graphs keep their order and action names. No option, API, wire or persisted-state change, so running jobs and master failover do not re-parse anything.
1.3 Performance / Side-Effect Analysis
Parse-time only, O((V + E) log V) with V as the number of transforms, nothing per record. No new threads, locks or retries. Nothing sensitive is logged.
1.4 Error Handling and Logging
JobDefineCheckException on the Zeta path and ConfigCheckException in dry-run, both with the cycle path or the unresolved IDs. No new findings.
2. Code Quality Assessment
2.1 Coding Standards
Good. Javadoc and comments on the scheduler are present.
2.2 Test Coverage and Test Stability
Strong and unchanged: ConfigParserUtilTest, MultipleTableJobConfigParserTest, SeaTunnelConfValidateCommandTest and JobExecutionIT#testRejectsExplicitTransformSelfCycleBeforeSubmission cover cycles, partial or wholly unresolved inputs, duplicate IDs, reverse-declared chains, fallbacks and index overflow. Hang regressions use assertTimeoutPreemptively. Stability rating: Stable.
CI for this head: the apache-side PR signal is success, and it is only a pointer. The fork Build run for 3696a4b3c07d (https://github.com/goutamadwant/seatunnel/actions/runs/39903819357) is still queued, so there is no result for this head yet. On the previous head the red unit-test legs came from JdbcOptionRuleCompletenessTest (numeric_special_value_as_null) in connector-jdbc, which stopped the reactor before engine-core, client and starter ran. This head contains #12800, and the test in the tree now references NUMERIC_SPECIAL_VALUE_AS_NULL explicitly, so I expect that break to be gone, but that is a reading of the code, not an observed run.
2.3 Documentation Updates
en and zh entries exist. See Issue 1 below for the one remaining sentence.
3. Architectural Soundness
3.1 Elegance
Long-term solution: one shared scheduler instead of three divergent loops.
3.2 Maintainability
Much better than before. The Fenwick bookkeeping is documented and pinned by index-boundary tests.
3.3 Extensibility
New fallbacks or dependency sources touch one class.
3.4 Historical-Version Compatibility
Only submission-time parsing changes. No serialized job, checkpoint or savepoint format is touched.
4. Issue Summary
Carried over from my previous review, both still open at this head and both non-blocking:
| Number | Issue | Location | Severity |
|---|---|---|---|
| 1 | Migration guide does not mention that a job resubmitted from a savepoint with an unresolved or partial plugin_input now fails at parse, and once fixed the previously dropped upstream feeds the transform |
docs/en and docs/zh incompatible-changes.md, "Transform Dependency Resolution" |
Low |
| 2 | No comment explains why the single-transform fallback depends on anyInputAvailable (a one-transform job with some existing input is still treated as a real mismatch) |
TransformDependencyScheduler.java:136-149 |
Low |
5. Merge Recommendation
Conclusion: Ready to merge
Blockers: none in source. The only remaining gate is a green Build on this head. Please let the fork run above finish and share the result; if only unrelated jobs are red, rerun just those. Please do not add unrelated patches to this PR.
Recommended non-blocking: Issues 1 and 2 above (one sentence in each migration guide, one short code comment).
Overall assessment: a careful, well-tested fix for a real hang and a silent data-lineage bug. The PR-owned code has not changed across the recent dev syncs. I have no source-level objection; I am holding the approval only because there is no CI result for this head yet. A maintainer with write access still has to do the final approval and merge.
|
Thanks for the rebase. Comparing Addressed
One remaining question
Otherwise this looks good to me. |
|
Thanks @SEZ9 for the careful re-check on the synced head. On F8, I re-read head F8 on this head:
Optional cleanup for @goutamadwant (not blocking): pass For completeness: the |
Signed-off-by: Goutam Adwant <workwithgoutam@gmail.com>
DanielLeens
left a comment
There was a problem hiding this comment.
Thanks @goutamadwant, and thank you for picking up both of my non-blocking items so quickly. The head moved to 9152a91fa278 after my last review at 3696a4b3c07d. It is one new commit ([Docs][Zeta] Explain restored transform-input migration) touching only the en and zh incompatible-changes.md and one comment in TransformDependencyScheduler.java. I still did a fresh full re-read of the diff and call chain at this head rather than only the delta. Analysis is static; nothing was built or run locally.
What Problem Does This PR Solve?
- User pain point: in Zeta job parsing and
--dry-run connect, a cyclic transform graph made the retry queue spin forever, and a multi-input transform was built as soon as any one of its configured inputs existed, silently dropping the other upstream edges (#12077). - Fix approach: one shared
TransformDependencySchedulerinengine-corereplaces the retry queues in the runtime parser, the graph pre-check and the dry-run validator. It releases each waiting set exactly once, keeps the legacy evaluation order and action indexes, detects cycles, and delays plugin discovery until every input exists. - One-sentence summary: cycles and unresolved inputs now fail fast with a concrete message, and valid graphs keep their order, names and indexes.
1. Code Change Review
1.1 Core Logic Analysis
What changed since 3696a4b3c07d (verified with the compare view): incompatible-changes.md (en, zh) and a 2-line comment above singleTransformLegacyFallback in TransformDependencyScheduler.java. No executable line changed, and all PR-owned source, test and E2E files are otherwise identical, so my earlier analysis of the code still holds:
Client: MultipleTableJobConfigParser.parse
-> ConfigParserUtil.checkGraph (cycle / unresolved pre-check)
-> parseTransforms -> TransformDependencyScheduler.scheduleTransforms -> parseTransform (plugin discovery only after readiness)
CLI: DryRunConnectValidator.validateTransforms -> same scheduler -> validateTransform
- Each output ID releases its waiters once, so duplicate output IDs cannot release a multi-input transform early.
- After the ready loop it checks cycles, then applies the two documented fallbacks, then fails with the unresolved
output <- inputspairs. - Dry-run and runtime share one resolution path, so they cannot disagree.
- This is the normal job-submission parse path (every job with transforms), not a recovery-only path.
1.2 Compatibility Impact
Partially incompatible by design, and documented in both languages: configs that relied on the old hang or on silently dropped inputs now fail at submission. The new migration text also states that the same validation runs when a job is resubmitted from a savepoint and that a corrected or reordered graph now feeds previously dropped upstreams to the transform, so users should review downstream output. That closes my earlier Issue 1. Valid graphs keep their order and action names. No option, API, wire or persisted-state change.
1.3 Performance / Side-Effect Analysis
Parse-time only, O((V + E) log V) in the number of transforms, nothing per record. No new threads, locks or retries.
1.4 Error Handling and Logging
JobDefineCheckException on the Zeta path and ConfigCheckException in dry-run, both with the cycle path or unresolved IDs. No new findings.
2. Code Quality Assessment
2.1 Coding Standards
Good. The new comment explains why a partially matched input list is a mismatch and not the single-transform fallback, which closes my earlier Issue 2.
2.2 Test Coverage and Test Stability
Strong and unchanged: ConfigParserUtilTest, MultipleTableJobConfigParserTest, SeaTunnelConfValidateCommandTest and JobExecutionIT#testRejectsExplicitTransformSelfCycleBeforeSubmission cover cycles, partial or wholly unresolved inputs, duplicate IDs, reverse-declared chains, fallbacks and index overflow. Hang regressions use assertTimeoutPreemptively. Stability rating: Stable.
2.3 Documentation Updates
en and zh entries exist and are in sync for this change.
3. Architectural Soundness
- 3.1 Long-term solution: one shared scheduler instead of three divergent loops.
- 3.2 Maintainability: much better than before; the Fenwick bookkeeping is documented and pinned by index-boundary tests.
- 3.3 Extensibility: new fallbacks or dependency sources touch one class.
- 3.4 Historical-version compatibility: only submission-time parsing changes; no serialized job, checkpoint or savepoint format is touched.
4. Issue Summary
No open code or documentation issues at this head. My two earlier Low items (migration guide wording, missing comment) are resolved.
| Number | Issue | Location | Severity |
|---|---|---|---|
| - | None open | - | - |
5. Merge Recommendation
Conclusion: Ready to merge
Blockers (must fix): none in source or docs.
Recommended fixes (non-blocking): none.
CI status of this head: the apache-side Build check (https://github.com/apache/seatunnel/runs/121071491816) is still pending and only points at the fork's run, so there is no green result for this head yet. My approval is held only on a green Build for 9152a91fa278; if a job fails, please share the failing link and I will help determine whether it is related to this diff, and then only that job should be rerun. The branch is 14 commits behind dev with no conflicts, so there is nothing to sync for correctness reasons.
Overall assessment: a careful, well-tested fix for a real hang and a silent data-lineage bug, with a clear migration note. I have no source-level objection. A maintainer with write access still has to do the final approval and merge.
DanielLeens
left a comment
There was a problem hiding this comment.
Thanks for the follow-up, @SEZ9 and @goutamadwant. I re-read the full fourteen-file diff and current call chains. The F1-F8 technical points are addressed as detailed below, and the two small items from my latest full review (savepoint migration wording and the partial-input fallback comment) are now fixed. Sorry for not calling out the new public resolver/wrapper documentation gap earlier: Issue 4 below is a carryover delivery omission, not a regression introduced by this documentation update.
What problem this PR solves
- User pain: Transform configuration parsing can loop forever on unresolved cycles or construct a multi-input transform from only the already available inputs, silently dropping intended upstreams.
- Fix approach: Use one indexed dependency scheduler in graph validation, runtime parsing and connect dry-run; wait for every distinct dependency, reject cycles/unresolved graphs, retain documented legacy fallbacks and preserve legacy action indexes.
- One-line summary: Resolve valid transform graphs deterministically and fail invalid graphs before plugin discovery instead of hanging or dropping inputs.
Example: For a transform requiring [sourceA, transformB], sourceA being ready is no longer enough: transformB must also be available before the factory receives its schemas.
1. Code-change review
1.1 Core logic analysis
Fourteen files, +1604/-81, including scheduler/parser, dry-run, UT/E2E and EN/ZH migration notes. The complete target-relative diff was read at 9152a91fa278e7827e66b73c691e6f1e8f961abc, not just the latest commit. Submission-time production parsing plus matching test/documentation changes.
Before / after:
Before: poll queue; filter unavailable inputs; accept partial input list or requeue indefinitely
After: scheduleTransforms -> missing distinct IDs -> ready set -> release waiters once
-> unresolved cycle check -> narrow fallback -> contextual failure
Before: discovery could happen before readiness
After: parseTransform checks readiness before getTransformPluginJarPaths
Complete relevant execution chain:
ClientJobExecutionEnvironment.getLogicalDag -> parser.parse
-> ConfigParserUtil.checkGraph():70-78 / complex phase :175-181
-> TransformDependencyScheduler.scheduleTransforms
-> MultipleTableJobConfigParser.parseTransforms():470-480
-> same scheduler -> scheduled actionIndex/config/fallback
-> parseTransform():492-529 -> all inputs present -> plugin discovery
-> transform factory -> produced tables/actions -> downstream sink
Connect dry-run -> DryRunConnectValidator.validateTransforms():211-225
-> scheduleTransforms wrapper :228-235 -> same scheduler
-> validateTransform :240-280 reads scheduled IDs -> schema/factory checks
Scheduler :59-83 missing ID index -> circular TreeSet ready selection
-> Fenwick remaining counts preserve queue-poll indexes :87-119
-> waitingByInputId.remove(outputId) :121-129 releases once per ID
-> unresolved check :132-157; cycle check before fallback
-> final omitted/empty input OR single-transform total mismatch fallback
-> otherwise JobDefineCheckException (translated to ConfigCheckException in dry-run)
Running/checkpoint recovery uses the submitted DAG; resubmission from savepoint
parses again and therefore applies the documented stricter validation
Key findings:
- +1 to @SEZ9's F1 requirement: removing the waiting list on first output availability prevents a duplicate producer from decrementing a join twice. Tests assert actual scheduling order and upstream edges, not just a final success.
- F2/F4/F6 are resolved: explicit self-cycles fail before fallback; omitted input first resolves DEFAULT_ID; only the documented terminal/one-transform exceptions remain.
- F3/F5: both callers use the same scheduler and ConfigParserUtil.getInputIds. Dry-run reads scheduled IDs directly. Runtime re-reads immutable config with the same helper/expression; redundant, but not a divergent resolver.
- F7: both migration notes now explicitly cover savepoint resubmission and previously dropped inputs becoming actual downstream inputs.
- F8: ScheduledTransform.getOutputId has a production caller at DryRunConnectValidator:240. It is not unused. Runtime parseTransform does not call that accessor; the earlier correction to that distinction still stands.
Community context: All SEZ9 discussion findings and their later resolutions were read. I agree with the latest source assessment, and provide concrete F8 closure rather than repeating the request. There is no non-Daniel formal approval to treat as an effective final repository gate.
1.2 Compatibility impact
Partially incompatible, intentionally documented. Valid graphs retain evaluation order/default action names. Cyclic/unresolved/partially bound configurations that previously hung or silently discarded inputs now fail during parsing/dry-run. No config key/default, wire format or persisted state format changes; exposing the existing input resolver is an additive API change. Migration: correct explicit input/output references and cycles before submission/resubmission; check downstream output when an intended upstream was previously dropped. EN/ZH incompatible-changes entries describe the retained fallbacks.
1.3 Performance / side effects
Parse-time CPU/memory changes only; no per-record cost, new runtime thread, lock, retry or resource lifecycle. Indexed missing dependencies plus TreeSet/Fenwick bookkeeping avoids replaying unsuccessful queue polls: O((V+E) log V) scheduling with O(V+E) bookkeeping. Graph pre-check and parser can invoke the scheduler separately, a bounded repeated parse-time cost. Long action indexes are range-checked before narrowing to int.
1.4 Error handling and logging
Unresolved/cycle failures identify output/input IDs and concrete cycles before plugin discovery. Dry-run translates the exception type while retaining the cause/message. Unknown upstream schemas remain explicitly skipped, not falsely validated. No credential values are introduced into the new messages.
Issue 3: Optionally pass scheduled IDs into the runtime parser
- Location:
MultipleTableJobConfigParser.java:470-507. - Problem: Runtime parseTransform reads the same immutable config IDs again, while dry-run uses ScheduledTransform accessors. This is the redundancy discussed under F8, not a correctness divergence.
- Risk: Small maintenance duplication if ID semantics change later.
- Best fix: Optionally pass the already resolved input/output IDs along with actionIndex/config/fallback, or retain an explicit note about the shared deterministic resolver.
- Severity: Low, non-blocking. Raised by another reviewer: Related to @SEZ9 F8; this adds the precise runtime-call-site distinction, rather than reasserting an unused getter.
Issue 4: Document the public ID resolver and dry-run scheduler boundary
- Location:
ConfigParserUtil.java:269-271;DryRunConnectValidator.java:228-235. - Problem: The now-public getInputIds method has no contract comment for omitted input versus an explicit empty list, and the new wrapper lacks a comment stating it translates scheduler failures while preserving the shared resolution semantics. Those distinctions control the compatibility fallbacks and are central to this change.
- Risk: Future callers can conflate omission with an empty list or reintroduce dry-run/runtime divergence while changing the wrapper.
- Best fix: Add concise method comments explaining the default-ID/explicit-empty contract and exception translation. No algorithm rewrite is needed.
- Severity: Medium, required core-method documentation under the review rules. Raised by another reviewer: No; carryover omission acknowledged above.
2. Code-quality assessment
2.1 Code conventions
The scheduler class/method, Fenwick intent, cycle detection and fallback rationale have comments; new files have ASF headers. Issue 4 covers the remaining newly exposed/core wrapper contracts. The scoped whitespace check is clean. The scoped whitespace check produced no entries.
2.2 Test coverage and stability
Stability rating: Stable. ConfigParserUtilTest uses seeded Random(12079L), bounded legacy-fixture polling and exact evaluation/index assertions; no wall-clock performance threshold is used. Parser/CLI tests have preemptive timeout guards for the former hang. Factory creation-order state is synchronized and explicitly reset before order-sensitive assertions. JobExecutionIT uses try-with-resources and a 30-second guard for pre-submission self-cycle rejection. No local build, unit/integration test, container or service was run. Coverage includes explicit/implicit self-reference, duplicate default/explicit outputs, partial inputs, reverse declarations, terminal omitted/empty fallback, actual schemas/edges, index overflow, and seeded legacy-order comparison. The E2E addition exercises getLogicalDag through a real client setup; no job should be submitted for its invalid graph.
Current exact-head fork Build is still queued, with no conclusion. The prior Build failure is not the current run result. This head contains merged JDBC fixture/UID fix #12938 (ancestor check exit 0). The newer empty-provider baseline fix #12895 is not an ancestor, but there is no current failing lane demonstrating that issue on this head, so I am not requesting a speculative sync or unrelated patch. Let this run start/finish, or inspect its queue/approval state if it remains queued.
2.3 Documentation
EN/ZH incompatible-changes notes are present at EN:179 and ZH:153 and now include the savepoint resubmission/output implications. Previous Issue 1 is closed. The explanatory comment around partially available inputs closes previous Issue 2.
3. Architectural soundness
3.1 Solution quality
A long-term shared resolution path fixes both a hang and silent lineage loss, with narrow legacy compatibility instead of divergent ad hoc loops.
3.2 Maintainability
Centralized scheduling, explicit fallback state and behavioral regressions are strong boundaries; finish the two method comments and consider removing the redundant runtime ID read later.
3.3 Extensibility
Keep shared options linked to their actual definitions and connector-specific limits explicit. No new extension API is introduced.
3.4 Historical compatibility
No runtime migration, state rewrite or dependency replacement is introduced; examples must match the version being used.
4. Findings summary
| Number | Finding | Location | Severity |
|---|---|---|---|
| 4 | Public resolver/wrapper contracts need comments | ConfigParserUtil.java:269; DryRunConnectValidator.java:228 | Medium, required |
| 3 | Optional reuse of scheduled IDs in runtime parser | MultipleTableJobConfigParser.java:470-507 | Low |
Previous Issues 1-2 and SEZ9 F1-F8 technical questions are addressed.
5. Merge conclusion
Conclusion: Can merge after fixes
- Blocking items: Issue 4's concise method documentation, then a completed current-head Build and effective write-capable maintainer approval. No algorithmic/source-behavior blocker was found.
- Non-blocking recommendations: Issue 3 is optional cleanup. Keep source/CI conclusions separate and do not add unrelated connector fixes merely to chase a queued run.
I agree with the core direction and SEZ9's resolved technical points. A small documentation follow-up is sufficient on the code side; an alternative scheduler is unnecessary. The current Build must complete before approval can be treated as merge-ready.
Compare against dev: diverged, ahead 19 / behind 14, merge conflict=False. Drift is recorded independently; it does not itself justify a sync-first comment without an unrelated CI failure.
Daniel has READ permission. This COMMENT is a technical assessment, not an effective write-level approval. A write-capable maintainer must confirm repository gates and supply the effective approval.
Local evidence: exact-head fetch/checkout on seatunnel-review-12079, complete diff and source call-chain reads cited above. No runtime verification was claimed.
|
Thanks for the follow-up on the F1–F8 points. Based on the latest automated pass at A few things I still want to confirm before approving, since the review summary I have is truncated and I'd rather see it in the PR than assume:
Once those are visible in the PR (tests + a short pointer to the doc sections), I'm happy to do the final pass. |
|
Hi @SEZ9 and @goutamadwant, thanks for the detailed questions. I re-read head
On CI: the Build for this head is a failure, but the one failing job I could inspect is So the only item between this PR and a clean source review is the two method comments, plus a green rerun. Thanks again for the careful work! |
|
Thanks for the detailed pointers on I can't mark the findings resolved from the comment alone, though. I'll go through the diff on this head and confirm the tests and docs changes directly before updating the review. Two things in the meantime:
Once I've verified against the diff and have those two clarifications, I expect this to be close to ready. |
|
Thanks @SEZ9, and for taking the time to go through the diff on this head (
I will re-verify once the Javadoc commit lands. The current |
Signed-off-by: Goutam Adwant <workwithgoutam@gmail.com>
Signed-off-by: Goutam Adwant <workwithgoutam@gmail.com>
DanielLeens
left a comment
There was a problem hiding this comment.
Thanks for adding the two method contracts. I re-read the full current diff at 67f3bde and the runtime/dry-run call chains. The previous required documentation finding is resolved, and I have no blocking production-source finding. Build remains in progress, with a completed connector-lane failure caused by a Docker Hub 502 during Spark image setup. One conditional test-isolation improvement and the earlier optional runtime-ID reuse remain non-blocking.
What Problem Does This PR Solve?
User pain point: Transform parsing could requeue unresolved dependencies forever, or construct a multi-input transform after only one upstream was available and silently omit other configured inputs.
Fix approach: Use a shared indexed scheduler for graph checks, runtime transform parsing and connect dry-run; wait for all distinct IDs, release each output ID once, diagnose cycles/unresolved dependencies and preserve the deliberately retained legacy bindings and action indexes.
One-sentence summary and example: A transform requiring sourceA and transformB now waits until both are available; an explicit self-cycle fails with a concrete cycle path rather than entering the legacy last-table fallback.
1. Code Change Review
1.1 Core Logic Analysis
The scheduler records missing distinct input IDs and ready/remaining config indexes. Circular TreeSet selection reproduces retry-queue evaluation order, while a Fenwick tree counts skipped remaining entries to preserve poll-derived action indexes without actually replaying them. Releasing waitingByInputId.remove(outputId) makes repeated producers satisfy an ID only once. After readiness is exhausted, Kahn cycle detection runs before the narrow terminal omitted/empty or single-transform total-mismatch fallback.
Before:
while (!remainingTransforms.isEmpty()) {
// Filter unavailable inputs; retry when none are ready.
}After:
List<ScheduledTransform> scheduled = scheduleTransforms(configs, availableOutputIds);
for (ScheduledTransform transform : scheduled) {
// Every declared dependency is available, or the explicit legacy fallback applies.
}Key Findings
- Distinct missing IDs prevent repeated configured inputs from over-counting, and removal of the waiting list prevents duplicate outputs from releasing a join early.
- Existing output values can be consumed before a same-ID overwrite, preserving in-place transform behavior.
- Omitted plugin_input first resolves DEFAULT_ID; an explicit empty list stays empty. Only one remaining omitted/empty transform may fall back, or a single transform with no matched input after cycle rejection. Partially matched input lists do not qualify.
- Cycle extraction is iterative and reports a concrete repeated path; it does not recurse with graph depth.
- Long action indexes are range-checked before narrowing to int.
- Runtime and dry-run deduplicate actual input actions/tables through linked sets while preserving their input order.
- The newly public getInputIds and dry-run wrapper now document omission/empty semantics and exception translation.
- Per-transform discovery/factory preparation follows readiness; the outer runtime parser still resolves connector jars/classloaders before its graph check, so this is not a claim that every bootstrap operation occurs after validation.
Complete runtime path
ClientJobExecutionEnvironment.getLogicalDag -> MultipleTableJobConfigParser.parse
-> ConfigParserUtil.checkGraph -> shared scheduler / existing graph checks
-> source discovery -> parseTransforms -> shared scheduler
-> parseTransform -> input actions/tables -> transform factory -> downstream sink
connect dry-run -> DryRunConnectValidator.validateTransforms
-> scheduleTransforms wrapper -> same scheduler -> scheduled IDs -> schema validation
savepoint resubmission -> configuration parse again -> same stricter validation
The dependency count tracks distinct unavailable IDs, while actual schemas are read in scheduled evaluation order, so overwrite behavior is not approximated by storing a schema in the scheduler. The circular remaining-index count reproduces unsuccessful legacy polls for valid graphs; it does not add a per-record cost. Cycles are checked before fallback, including a single explicit self-reference. Existing complex-graph explicit-ID and linkage requirements remain ahead of the shared cycle phase; direct parseTransforms/dry-run acceptance does not imply every omitted-ID graph passes the complete runtime graph validator.
1.2 Compatibility Impact
Partially incompatible. Previously cyclic/unresolved or partially bound configurations now fail instead of hanging or silently dropping upstreams. The two documented fallback families remain. Correctly resolved valid graphs retain evaluation order and default transform action names. Previously dropped upstreams can now affect output after a corrected or valid out-of-order graph is parsed, including savepoint resubmission. No job option, connector registration, serialized checkpoint format or wire protocol is removed.
1.3 Performance / Side-Effect Analysis
Scheduling is O((V+E) log V) with O(V+E) bookkeeping rather than repeated queue polling. The graph pre-check and runtime parse may each schedule once, a bounded parse-time cost. Kahn processing and iterative cycle extraction avoid graph-depth recursion. There are no added runtime threads, locks, network calls or per-record allocations. Very large poll-derived indexes fail explicitly rather than overflow.
1.4 Error Handling and Logging
Runtime throws JobDefineCheckException with unresolved output/input pairs or a concrete cycle. Dry-run translates to ConfigCheckException preserving message and cause, then retains existing contextual/sanitized reporting. Unknown upstream schemas remain SKIPPED rather than falsely validated. Missing scheduled inputs have a defensive runtime assertion before transform factory work.
Issue 1: Isolate the shared transform-creation recorder in tests
- Location:
seatunnel-core/seatunnel-starter/src/test/java/org/apache/seatunnel/core/starter/seatunnel/command/DryRunTestTransformFactory.java: CREATED_TABLES; SeaTunnelConfValidateCommandTest order assertions - Problem description: The test factory stores creation order in one static list. Individual add/reset/snapshot operations are synchronized, but a test's reset -> validation -> assertion sequence is not isolated from another test using the same factory.
- Potential risk: If these methods are run concurrently, another test can clear or append to the recorder and make the exact-order assertions fail. No enabled parallel mode for this starter suite or observed CI failure from this mechanism was found; this is a conditional fixture risk, not a production defect.
- Best improvement: Use a per-invocation recorder/correlation, or a shared test resource lock covering every use of this factory. Retain the actual order/schema assertions.
- Severity: Medium; non-blocking.
- Raised by another reviewer: No
Issue 2: Optional reuse of scheduled IDs in runtime parsing
- Location:
seatunnel-engine/seatunnel-engine-core/src/main/java/org/apache/seatunnel/engine/core/parse/MultipleTableJobConfigParser.java: parseTransforms/parseTransform - Problem description: Runtime passes the scheduled config/index/fallback but reads the input/output IDs again from immutable config; dry-run consumes the scheduled IDs directly. Both input reads use the same getInputIds helper, so this is currently redundant rather than inconsistent.
- Potential risk: Small maintenance duplication if future ID resolution semantics change.
- Best improvement: Optionally pass the scheduled input/output IDs into parseTransform in a focused cleanup; no scheduler rewrite is required.
- Severity: Low; non-blocking.
- Raised by another reviewer: Continuation of the earlier Daniel/SEZ9 F8 discussion; not an unused-getOutputId finding.
2. Code Quality Assessment
2.1 Coding Standards
The scheduler, Fenwick poll-count intent, cycle path extraction and fallback distinctions are explained in comments. The new public resolver and dry-run translation contracts now close the previous Medium documentation blocker. ASF headers and test-only factory service registration are present.
2.2 Test Coverage and Test Stability
Test stability: Risk present. Issue 1 records the conditional static-recorder isolation risk; the previous blanket Stable description was too broad for parallel reuse. Core tests use seeded Random(12079L), bounded legacy queue simulation, exact order/index assertions and an explicit integer-boundary case rather than a fragile speed threshold. Parser and dry-run tests assert actual schemas/edges, duplicate default outputs, reverse declarations, missing/partial inputs, omitted-vs-empty binding and cycle message text. Timeout guards bound former-hang checks. The added JobExecutionIT opens/closes a real client and checks getLogicalDag rejects the self-cycle before submission; it does not execute a successful transform job or prove checkpoint recovery. No local tests, containers or jobs were run.
Local analysis: fetched the head, created independent branch seatunnel-review-12079, read the complete 14-file diff and traced affected callers with git show/git grep. No local compilation, unit tests, integration tests, E2E, containers, or job execution was performed, as requested. These are source-analysis conclusions; execution evidence is the current-head CI below.
2.3 Documentation Updates
English and Chinese migration entries describe strict failure, retained fallbacks, explicit-ID requirements, savepoint resubmission and changed downstream output when previously omitted inputs become connected. Required developer-facing method contracts are now present as well.
3. Architectural Soundness
3.1 Elegance of the Solution
Precise fix. Dependency readiness belongs in one engine-core scheduler shared by runtime parsing and dry-run, rather than separate retry-loop patches. The scheduler resolves order/binding, while callers retain schema/plugin responsibilities. The smallest safe fix and the preferred architectural boundary coincide. A wholesale DAG/state redesign would add unnecessary restore and compatibility risk.
3.2 Maintainability
The scheduler centralizes dependency and fallback policy. Fenwick bookkeeping is additional complexity justified by preserving legacy poll-based action names without quadratic retry work, with deterministic legacy-equivalence tests. Keep input-read cleanup optional and independent of correctness.
3.3 Extensibility
Future dependency sources or fallback policies must be added at this shared boundary and checked through both actual parser and dry-run schema tests. Keep semantic validation separate from merely naming an output ID.
3.4 Historical-Version Compatibility
The submitted running DAG and checkpoint representation are not rewritten. Resubmission from a savepoint reparses config and therefore applies the stricter behavior; the migration notes now state that. This source review does not claim an executed old-version restore matrix.
4. Issue Summary
| Number | Issue | Location | Severity |
|---|---|---|---|
| 1 | Isolate the shared transform-creation recorder in tests | seatunnel-core/seatunnel-starter/src/test/java/org/apache/seatunnel/core/starter/seatunnel/command/DryRunTestTransformFactory.java: CREATED_TABLES; SeaTunnelConfValidateCommandTest order assertions |
Medium |
| 2 | Optional reuse of scheduled IDs in runtime parsing | seatunnel-engine/seatunnel-engine-core/src/main/java/org/apache/seatunnel/engine/core/parse/MultipleTableJobConfigParser.java: parseTransforms/parseTransform |
Low |
5. Merge Recommendation
Conclusion: Ready to merge after fixes
- Blockers — must be fixed
- No source blocker identified. No blocking production-source finding remains. Obtain green current-head CI and effective maintainer approval. The numbered follow-ups are non-blocking.
- Recommended fixes — non-blocking
- Issue 1: Isolate the shared transform-creation recorder in tests; Issue 2: Optional reuse of scheduled IDs in runtime parsing
Overall assessment: The shared scheduler fixes the hang and partial-input loss while deliberately retaining supported legacy behavior. The previous required comment fix is complete. CI is still a gate even though the diagnosed connector failure occurs in image setup.
Existing-review reconciliation: +1 to SEZ9's F1-F8 checks. F1 is covered by one-time waiting-list removal plus ConfigParserUtilTest.testDuplicateOutputsReleaseEachMissingInputOnlyOnce, MultipleTableJobConfigParserTest.testDuplicateUnnamedOutputsWaitForOtherJoinInput and the matching dry-run test. F2/F4 are covered by terminal omitted/default binding tests and migration notes. F3/F5 converge on TransformDependencyScheduler and ConfigParserUtil.getInputIds. F6 dry-run explicitly asserts self -> self, matching runtime coverage. F7 now includes both migration and method contracts. F8 getOutputId has a production dry-run caller; runtime ID rereading is the optional Issue 2, not an unused API. Historical approval events are on earlier commits; no current effective independent approval is inferred. Author replies have been incorporated into this single full review.
Head 67f3bde2506aef2e6d0b3456f76e7ff59ad5932e. Live base dev at 3ac2d1cec6aa7b3741c4dacdd23020f778e23c02; comparison ahead, ahead 21 / behind 0; merge state BLOCKED. Behind/diverged alone is recorded metadata, not a request to synchronize upstream. Checks: Signal pull request update = SUCCESS; Build = IN_PROGRESS
Current fork run 41021951825 is still active. all-connectors-it-4, JDK 11 completed with one StarRocksIT.testStarRocksReadRowCount parameter-resolution error: Docker could not fetch a tyrantlucifer/spark image layer from registry-1.docker.io because it returned 502 Bad Gateway. This is before the test body/config parser; the relevant StarRocks/common fixture and workflow have no PR-owned diff. No verified upstream-sync cure is established. After checking the active run/attempt state, retry only that failed lane if still required; do not patch scheduler code or start an overlapping whole-workflow rerun. No rerun was triggered here. Aggregate Build remains IN_PROGRESS rather than green.
Daniel has READ repository permission. This COMMENT records the technical conclusion; it is not an effective maintainer approval or a dismissal of any historical requested-changes entry. A write-capable maintainer owns the final repository approval gate.
|
Thanks for the follow-up on 67f3bde. Going through the earlier points against what is described so far:
Concrete asks before I approve: (1) diff/test pointers for F1 and F6, (2) the multi-transform omitted- |
Purpose of this pull request
Closes #12077.
This PR fixes transform dependency resolution in the Zeta runtime parser and connect dry-run validator.
The previous retry queue could loop indefinitely on a dependency cycle. It also considered a multi-input transform ready when only some configured input IDs were available, silently removing unresolved upstream edges from the generated DAG. Reverse-declared valid chains repeatedly retried transforms and performed plugin discovery before readiness was known.
The change:
O((V + E) log V)time andO(V + E)memory;No public API, configuration option, dependency, or default behavior is added.
Does this PR introduce any user-facing change?
Yes.
Previously, a cyclic transform graph could leave submission or
--dry-run connectrunning indefinitely. The same resolver could create an out-of-order multi-input transform from only the inputs already available, silently dropping the remaining configured upstreams.After this change, cycles and unresolved dependencies fail promptly with a contextual configuration error. Valid out-of-order DAGs wait for all inputs and retain every upstream action/schema. Correctly ordered jobs, duplicate input IDs, generated action names, result order, and the existing compatibility fallbacks remain unchanged.
How was this patch tested?
Earlier validation on both supported JDK lines (historical results, not a fresh execution of the September 21 evidence review):
1.8.0_172):ConfigParserUtilTest9/9,MultipleTableJobConfigParserTest21/21, andSeaTunnelConfValidateCommandTest29/29 passed.11.0.19): the same suites passed 9/9, 21/21, and 29/29.seatunnel-engine-core,seatunnel-engine-client, andseatunnel-starter, including Spotless, compilation, and packaging.git diff --checkpassed.Coverage includes two-node and self cycles, partially and wholly unresolved inputs, mixed source/transform inputs, duplicate input IDs, valid reverse-declared chains, runtime and connect dry-run paths, simple and explicit-empty fallback compatibility, terminal sink selection, and action-index boundaries.
A wider aggregate reactor attempt was not counted as passing because it stopped before the affected modules in unchanged
seatunnel-config-shadegenerated sources with missingConfigException/ConfigOriginsymbols. The affected three-module verification above passes on both Java versions.Current review evidence
The runtime parser and connect dry-run use the same
TransformDependencySchedulerin engine-core. Waiting lists are consumed once withwaitingByInputId.remove, so duplicate output producers cannot satisfy another required input. Explicit self-cycles are rejected before legacy omitted-input fallback. The current regression suite includes duplicate unnamed outputs, terminal omitted-input schema binding, default-first binding and single explicit self-cycles.Compatibility and migration notes: English and Chinese.
The September 21 follow-up is a source/test evidence audit, not another test run or a claim of green aggregate CI. The optional getOutputId cleanup remains a separate follow-up, not an implemented part of this PR.
Check list
incompatible-changes.mdnotes are included; supported valid DAG ordering, generated names and compatibility fallbacks are preserved.