DanielLeens commented on PR #11557:
URL: https://github.com/apache/seatunnel/pull/11557#issuecomment-5289496990
**[Continuing the full review from my previous comment above — this is Part
2 of 2.]**
### Issue 33 — NEW. `checkpointAssignmentProofs` and
`uncheckpointedAssignmentSplitIds` are unbounded
**Location:** `ManagedSourceReaderRuntime.java:109-110`, against `:888-892,
:901, :909-916`
**Problem description:** `completedCheckpoints`/`abortedCheckpoints` are
explicitly capped at `MAX_CHECKPOINT_HISTORY = 128`, but these two collections
are pruned only when a `CHECKPOINT_COMPLETE`/`CHECKPOINT_ABORTED` command
actually arrives.
**Potential risk:** If the coordinator stops delivering completion
notifications — which is exactly what happens when checkpointing is degraded,
and exactly what Issue 1 causes — the map accumulates one `LinkedHashSet` of
every split id per checkpoint, forever. OOM on a long-running high-split-count
source, precisely when the job is already degraded.
**Best improvement:** Bound `checkpointAssignmentProofs` to
`MAX_CHECKPOINT_HISTORY` entries, evicting the lowest checkpoint id and folding
its ids out of `uncheckpointedAssignmentSplitIds`.
**Severity:** Medium
**Raised by another reviewer:** No
---
### Issue 34 — NEW. `subscribeToAvailability()` accumulates handlers on a
reused future
**Location:** `ManagedSourceReaderRuntime.java:641-666`, against `:763`
**Problem description:** Every `NOTHING_AVAILABLE` poll calls
`available.whenComplete(...)`. Because `applySplits` sets
`availabilitySignalled = true` (`:763`), a split assignment can re-open the
gate while an earlier subscription is still outstanding, producing a second
registration. Connectors commonly return the *same* `CompletableFuture`
instance from `isAvailable()` until it completes.
**Potential risk:** Unbounded retained-handler growth, each closure
capturing `this`, in any source that receives splits faster than it drains
them. A slow heap leak with no bound and no metric.
**Best improvement:** Track an `availabilitySubscribed` flag and skip
re-subscription while one is outstanding.
**Severity:** Medium
**Raised by another reviewer:** No
---
### Issue 35 — NEW. Async-task timeout cancellation leaks the interrupt flag
onto shared worker-pool threads
**Location:** `ManagedCoordinatorScheduler.java:337-395` (specifically
`:352-367`, `:383`); `TaskExecutionService.java:1121-1127, 1148-1167`
**Problem description:** On timeout, the scheduler calls
`worker.cancel(true)` (`:383`), interrupting whichever
`managedSourceIoExecutor`/`managedSourceCpuExecutor` thread is running the
connector callable. `FutureTask.handlePossibleCancellationInterrupt`
deliberately does **not** clear the interrupt status (the
`Thread.interrupted()` call is commented out in the JDK), and the scheduler's
own wrapper `finally` (`:362-366`) restores the context classloader but does
not clear the flag either.
**Potential risk:** These are node-wide shared pools. The next task
submitted to that thread — belonging to a *different* managed Source, possibly
a different job — begins with the interrupt flag set and fails spuriously. Same
defect class as Issue 7, in a second location, with cross-job blast radius.
**Best improvement:** Clear `Thread.interrupted()` in the wrapper's
`finally` at `:362-366`.
**Severity:** Medium
**Raised by another reviewer:** No
---
### Issue 36 — NEW. Kafka's managed split-back handler calls `assignSplit()`
eagerly, racing reader readiness
**Location:** `KafkaSourceSplitEnumerator.java:635-650`, `:510-511`, against
`ManagedSourceCoordinatorRuntime.java:1500-1507`
**Problem description:** The managed split-back completion handler calls
`assignSplit()` eagerly. The legacy path deliberately only parked splits and
let `registerReader` drive re-assignment. Both dispatch routes now hard-require
a ready reader via `requireReadyReader`. During reader-failure recovery a
reader is registered but `ready == false`, so the
`registeredReaders().isEmpty()` guard in `SeaTunnelSplitEnumeratorContext` does
not save it.
**Potential risk:** Two failure modes. First, `requireReadyReader` throws
and fails the job — and because it is a race between the admin round trip and
the epoch handshake, it presents as intermittent. In batch mode
(`discoveryIntervalMillis <= 0`, the default) `signalNoMoreSplits` is invoked
for every subtask, so any one not-yet-ready reader trips it. Second, and worse,
if all readers are momentarily unregistered, `assignSplit()` still runs
`assignedSplit.putAll(pendingSplit); pendingSplit.clear();` (`:510-511`),
permanently excluding those partitions from future discovery snapshots —
**silent partition loss with no error at all.**
**Best improvement:** Drop the `assignSplit();` call from the split-back
completion handler and let reader registration drive re-assignment, as the
legacy path does. This is also a concrete instance of Issue 31, so the general
deferral fix would mitigate the first failure mode but not the second.
**Severity:** High
**Raised by another reviewer:** No
---
### Issue 37 — NEW. `SourceReaderBase.isAvailable()` short-circuits on
`noMoreSplitsAssignment`, defeating event-driven availability
**Location:** `SourceReaderBase.java:170-182` (specifically `:172`)
**Problem description:** `if (currentFetch != null ||
!elementsQueue.isEmpty() || noMoreSplitsAssignment) return
CompletableFuture.completedFuture(null);` — the third term makes availability
**permanently complete** once no-more-splits arrives.
**Potential risk:** For an unbounded source in that state — Kafka with
static assignment, or any streaming source whose enumerator signalled
no-more-splits — the steady state becomes `pollNextManaged` ->
`NOTHING_AVAILABLE` -> `subscribeToAvailability()` -> already-complete future
fires `whenComplete` inline -> `availabilitySignalled.set(true)` -> poll again.
This is not a CPU burn, because `SourceFlowLifeCycle:354-356` sleeps
`idleWaitMillis` when `runOneTurn()` returns false — but it fully defeats the
event-driven availability mechanism, degrading it to a fixed 100 Hz poll per
subtask regardless of whether data exists.
**Best improvement:** Drop `noMoreSplitsAssignment` from the shortcut, or
qualify it with `Boundedness.BOUNDED`.
**Severity:** Medium
**Raised by another reviewer:** No
---
### Issue 38 — NEW. `FakeSourceReader.pollNextManaged` throws on a short
read, producing a permanent crash loop
**Location:** `FakeSourceReader.java:140-153`, with `:191-195`
**Problem description:** The managed poll loop throws
`IllegalStateException("Managed Fake source generator made no progress ...")`
whenever `generateFakedRows` returns 0, which it does whenever `startInclusive
>= effectiveRowCount(...)`.
**Potential risk:** Configure an explicit `rows` list of 5; the job
checkpoints mid-split at `(rowNum=3, totalRowNum=5)`; the config is then edited
to 2 rows and the job restored. `addSplits:194` keeps `(3, 5)` because neither
guard fires; `pollNextManaged` computes `start = 5 - 3 = 2`;
`effectiveRowCount` now returns 2; `endExclusive = min(2, 3) = 2`, the loop
body never executes, 0 is returned, the exception fires, the task fails, the
same state restores — **permanent crash loop.**
**Best improvement:** Treat a short read as end-of-split and poll the split
off, rather than throwing.
**Related Fake findings (Low):** `split_read_interval` now throttles only at
split start on the managed lane, gated on `split.getRowNum() ==
split.getTotalRowNum()` (`:128`), where legacy throttled every poll (`:86-90`)
— defensible, but an undocumented user-facing behaviour change. `addSplits`
(`:191-195`) dereferences `fakeDataGeneratorMap.get(split.getTableId())` with
no null check, moving a pre-existing NPE from poll time to assignment time.
Explicitly **not** a defect: the `addSplits` rewrite of `rowNum` to
`effectiveRowCount(...)` looks alarming but is inert on the legacy lane,
because `FakeDataGenerator.generateFakedRows` (`:122-138`) ignores `rowNum`
entirely when `fakeRows != null`.
**Severity:** Medium
**Raised by another reviewer:** No
---
### Issue 19 (rewritten) — Managed checkpoint-ack failure is propagated but
never logged, and can be lost outright
**Location:** `SeaTunnelTask.java:483-501, 505-520`;
`SourceSeaTunnelTask.java:192-205`; `SourceFlowLifeCycle.java:846-856`;
`ManagedSourceCoordinatorRuntime.java:936-951, 1597-1599`
**Problem description (corrected):** My prior claim that the failure is
thrown inside `whenComplete` and swallowed is **wrong**.
`waitForCheckpointAcknowledge()` defaults to `true`, so legacy tasks keep exact
`.join()` semantics, and the managed override routes the failure into
`reportManagedRuntimeFailure` -> `reportAsynchronousFailure`, consumed by the
owner loop at `:398`. That is a legitimate path. Three smaller real defects
remain:
**(19a)** `SourceSeaTunnelTask.onCheckpointAcknowledgeFailure` (`:198-205`)
replaces the base body entirely, so the base class's `log.error(...)`
(`SeaTunnelTask.java:516-520`) never runs on the managed lane. Combined with
the zero-logging problem, a failed managed checkpoint ack produces no log line
at all.
**(19b)** `reportManagedRuntimeFailure` (`SourceFlowLifeCycle.java:846-852`)
guards on `isManagedReaderRuntime()`, which reads only `runtimeSelection` and
stays true for the task's whole lifetime, while `managedReaderRuntime` is
assigned late in `setCollector` and cleared conceptually at close. A late ack
failure arriving after `close()` sets `asynchronousFailure` on a runtime whose
owner loop has exited — the failure is lost with no logging.
**(19c)** The coordinator does the identical non-joined `whenComplete` at
`:936-951`, and its `reportAsynchronousFailure` (`:1597-1599`) is a bare
`compareAndSet` with no logging and no wakeup, unlike the reader's version.
Ack *ordering* across barriers is also no longer serialized on the managed
lane; acks are keyed by checkpoint id so `CheckpointCoordinator` likely
tolerates it, but it is an undocumented behavioural change.
**Best improvement:** Call `super.onCheckpointAcknowledgeFailure(...)` first
in the override; log unconditionally in `reportManagedRuntimeFailure` before
the guards; add logging and a wakeup to the coordinator's
`reportAsynchronousFailure`; document the ordering change.
**Severity:** Medium
**Raised by another reviewer:** No
---
### Issue 16 (rewritten) — Three non-failure-atomic release paths leak the
node-wide memory budget permanently
**Location:** `ManagedSourceReaderRuntime.java:1246-1262` (16a),
`ReaderCommandMailbox.java:278-294` (16b),
`ManagedSourceReaderRuntime.java:506-537` (16c)
**Problem description (corrected):** `ManagedSourceMemoryBudget` has no
`close()` and its `release()` validates before mutating, so it *is*
failure-atomic — my prior claim was wrong. The leak is real but lives
elsewhere. **(16a)** `releaseOutbound` mutates all six accounting fields, then
throws the underflow guard, leaving `outboundMemoryBudget.release(bytes)` at
`:1261` unreachable. **(16b)** `ReaderCommandMailbox.close()` sets `closed =
true` first, then runs a release loop that can throw mid-way, leaving the
remaining bytes unreturned and `commands` uncleared. **(16c)**
`ManagedSourceReaderRuntime.close()` has no `try/finally`, so a throw at `:522`
skips `ControlEvent::discard` (which releases the `admitSerializedLegacySplits`
reservation), `releaseAllOutbound()` and `lifecycle.closed()`.
**Potential risk:** The budget is a `TaskExecutionService` singleton, so
leaked bytes shrink the pool for **every other managed Source on the worker**,
permanently, until node restart.
**Best improvement:** Move `outboundMemoryBudget.release(bytes)` above the
underflow guard; wrap the mailbox release loop and each stage of `close()` in
`try/finally`.
**Severity:** Medium-High
**Raised by another reviewer:** No
---
### Issue 28 (rewritten) — narrow residual only
**Location:** `ManagedSourceReaderRuntime.java:441, 465`;
`SourceFlowLifeCycle.java:285-295, 438-442`
**Problem description (corrected):** My prior claim that the managed lane
never sets `SourceFlowLifeCycle.prepareClose` is **largely refuted** —
`:439-442` sets it on the managed lane. The residual is narrow: the
poll-status-driven path calls the *runtime's* `signalNoMoreElement()`
(`ManagedSourceReaderRuntime:441` -> `:465`), not the lifecycle's, so on that
path `prepareClose` stays `false` and the flush timer keeps admitting flush
signals to a drained reader (`SourceFlowLifeCycle:285-295`), burning reserved
mailbox capacity. The connector-driven path via
`SourceReaderContext.signalNoMoreElement()` is correct.
**Severity:** Low
**Raised by another reviewer:** No
---
### Issues 10-15, 17, 18, 20-25, 27, 29, 30 — carried forward
None of the underlying files changed since my last full read, so I am not
re-deriving unchanged analysis a fifth time. Locations for traceability, with
this round's verification status:
| # | Issue | Location | Verified this round |
|---|---|---|---|
| 10 | `coordinatorEpoch`/`coordinatorAttemptId`/`closed` read cross-thread
without `volatile`; a stale read returns `STALE_TARGET` and can kill the job
right after the handshake | `ManagedSourceReaderRuntime.java:114-115, 136, 235,
256-259, 510, 605, 747-748, 1429` | Yes, re-derived |
| 11 | Deferred split request removed before the readiness test, so it is
silently dropped | `ManagedSourceCoordinatorRuntime.java:1410-1419` | Yes,
re-read |
| 12 | `enumerator.run()` runs inline on the drain thread, blocking every
queued barrier for its duration; strictly worse than legacy |
`SourceSplitEnumeratorTask.java:445-449` | Yes, re-read |
| 13 | Kafka managed discovery fails the job on transient errors;
`scheduleNextManagedDiscovery()` (`:601`) sits **past** the `throw` with no
`finally`, so any log-and-continue fix that does not also move the re-arm
converts a job kill into silent permanent loss of partition discovery |
`KafkaSourceSplitEnumerator.java:584-601` | Yes, re-derived and expanded — High
|
| 14 | **Refuted as stated.** `endOffsetsTimestamp` is set only in `case
TIMESTAMP:` (`KafkaSourceConfig.java:206-232`), so it is null elsewhere. Real
residual: the managed lane applies timestamp end-offsets on
**dynamic-discovery** rounds too, where `fetchPendingPartitionSplit` never did,
so a partition discovered mid-job now stops at the end timestamp instead of
latest — arguably more correct, but unannounced |
`KafkaSourceSplitEnumerator.java:749-767` | Corrected — Low |
| 15 | `restoreMetadata()` has no owner check and no idempotence guard; a
re-delivered restore forces a RUNNING reader back to RESTORING permanently |
`ManagedSourceReaderRuntime.java:194`; `ManagedSourceLifecycle.java:187` | Yes,
re-derived |
| 17 | `markRestoredSplitsIncluded` uses `containsAll`, so a chunk in which
the reader already finished some splits is never reconciled and is replayed
whole on failover | `SourceAssignmentTracker.java:257-267`; callers `:283, 325,
876` | Yes, re-derived; duplicates, not loss |
| 18 | `transportRetries` / `transportFirstAttemptNanos` leak on every
reader replacement | `ManagedSourceCoordinatorRuntime.java:1172-1196` | Carried
forward |
| 20 | Per-turn `updateMetrics()` holds `admissionLock` while taking three
mailbox locks and scanning every buffered command |
`ManagedSourceReaderRuntime.java:392-393, 1512-1515` | Yes, re-derived |
| 21 | Epoch rebase unimplemented — a second epoch hard-throws;
`resetForEpoch`/`awaitSignal` have zero callers; `restoredAppliedWatermark` is
stored and never read; idleness is a fixed sleep |
`ManagedSourceReaderRuntime.java:741-748, 679, 1290`;
`ReaderCommandMailbox.java:215, 244`;
`ManagedSourceCoordinatorRuntime.java:706, 1812` | Yes, re-derived and expanded
|
| 22 | **Partly refuted.** The connector's `close()` (`:285-301`) does
discard both `callAsync` handles, but the engine closes the scheduler first, so
in-flight discovery is interrupted before `adminClient.close()`. Residual
window is small: `cancel(true)` interrupts without joining, so a late
`listOffsets` can fire post-close, get a `TimeoutException`, and be discarded.
Log noise, no corruption. Note the **legacy** branch of the same method closes
the admin client before `executor.shutdownNow()` — a genuine ordering bug that
pre-dates this PR | `KafkaSourceSplitEnumerator.java:285-301` | Corrected — Low
|
| 23 | **Premise refuted, real defect relocated.** No CDC connector can be
selected into the managed lane (selection is allowlist plus explicit
capability; CDC declares none). The real defect:
`SourceReaderBase.pollNextManaged` is **concrete**, so the engine's safety
check `if (isManagedReaderRuntime() && !(reader instanceof
ManagedSourceReader))` (`SourceFlowLifeCycle.java:222`) is now vacuous for the
entire `SourceReaderBase` family, including `IncrementalSourceReader` — whose
`pollNext` (`:103-120`) sends the split request that `pollNextManaged` never
sends, and which deliberately drops the `Boundedness.BOUNDED` check that
`SourceReaderBase:143` re-imposes. A CDC connector allowlisted in future would
hang with zero assignments, past a guard that reports it as safe. Fix: make
`pollNextManaged` abstract, or have `activateManagedRuntime()` reject
subclasses that override `pollNext` without overriding `pollNextManaged` |
`SourceReaderBase.java:61,133-167`; `IncrementalSourceReade
r.java:103-120`; `SourceFlowLifeCycle.java:222` | Corrected — Medium now, High
on first CDC onboarding |
| 24 | Metric names embed the per-attempt `executionId` under a
"bounded-cardinality" Javadoc, and the code runs on the legacy lane |
`SourceRuntimeMetrics.java:42-43, 151-160, 207-213`;
`ManagedSourceRuntimeMetrics.java:313-322` | Yes, re-derived |
| 25 | `SourceConfig` gains five fields with no `serialVersionUID`; no
`incompatible-changes.md` entry | `SourceConfig.java:26-30, 41-42` | Yes,
re-derived |
| 27 | E2E lane assertion is order-dependent on an accumulating shared
container log and passes vacuously | `ManagedSourceRuntimeIT.java:58-65, 72-78,
102-104` | Yes, re-derived |
| 29 | `requestSplit()` and `signalNoMoreElement()` throw before the epoch
handshake, breaking the standard `sendSplitRequest()`-in-`open()` pattern;
`checkOwner()` also rejects connector fetcher threads |
`ManagedSourceReaderRuntime.java:448-458, 465-479, 1075-1078`;
`SourceFlowLifeCycle.java:265, 271` | Yes, re-derived and expanded |
| 30 | Grouped low-severity items — see below | — | Partly re-verified |
**Issue 30, grouped Low items, with this round's additions:**
*Dead code.* `ManagedSourceRuntimeMode.MANAGED_READER` can never be selected
— `ManagedSourceCapability.validate()` rejects `supportsManagedReader &&
!supportsManagedCoordinator` (`:155-158`), which is the exact condition
`ManagedSourceRuntimeSelector:81-82` uses to pick it. Verified this round.
`admittedNanos` is written to the wire, always `0` at send time, and
unconditionally overwritten on receipt — and `System.nanoTime()` is not
comparable across JVMs anyway. The deprecated 4-arg `TaskExecutionContext`
constructor silently yields `executionId = 0`, defeating attempt fencing; it
now has zero callers.
*Wire hygiene.* `SourceCommandCodec` writes magic plus `PAYLOAD_VERSION` for
split assignment (`:64-65`) and barriers (`:104-105`) but not for
`encodeCheckpointId`, `encodeCommandApplied`, `encodeReaderCheckpointReport`,
`encodeRestoredSplits` or `encodeNoMoreSplits`. The envelope CRC32 covers the
payload only, not the header. `SourceCommandEnvelope implements Serializable`
with no `serialVersionUID` despite a Javadoc positioning it as explicitly
non-Java-serialized. `SplitIdChunkAccumulator.splitIds()` silently
de-duplicates across chunks via `LinkedHashSet`, absorbing what would be a
coordinator bug.
*Error handling.* `validateRestoredSelection` (`:1629-1641`) reports
"capability or lane mismatch" without saying which of the four fields differed
or what the expected and actual values were — the single most likely
operator-facing failure in the whole feature, with the least useful message.
Over-broad `catch (IllegalArgumentException)` in three serializers wraps an
entire method body and mislabels any nested IAE. `handleSchemaSignals()` runs
unconditionally after every poll and calls `lifecycle.beginSchemaChange`, which
requires `RUNNING`, so a schema-change marker in a final poll turns a graceful
drain into a failure. A non-snapshot barrier calls `finishBarrier` without a
matching `beginCheckpointBarrier` (currently harmless, latent).
*Config and style.* Two `ArrayBlockingQueue(4096)` plus a
`ScheduledThreadPoolExecutor` are allocated on every worker regardless of the
feature flag. Twelve YAML keys use bare `Long.parseLong` instead of the
key-labelled helper, so a typo yields an unattributed `NumberFormatException`.
An unknown key under `managed-source-runtime` is only warned about, so a
misspelled `connector-allowlist` silently leaves the feature off.
`seatunnel.yaml:46` describes the feature as "Source **serialization**" rather
than Source runtime. Inline fully-qualified type names
(`ManagedSourceCoordinatorRuntime.java:1630, 1664`,
`ManagedCoordinatorScheduler.java:377`) instead of imports, against repo
convention.
*Selector hardening.* `runtime-protocol-version` is validated only as
positive and only against the connector's declaration, never against
`ManagedSourceCapability.CURRENT_RUNTIME_PROTOCOL_VERSION` — so an operator can
configure `7`, a connector can declare `7`, and the engine will run v1 wire
code under a v1-hardcoded error message. Allowlisting a connector that declares
no managed capability hard-throws at `PhysicalPlanGenerator` for **every** job
using that connector, new submissions and restores alike; falling back to
legacy with a `WARN` would match operator intent better.
---
### Issue 26 — Test coverage does not reach the code that holds the risk,
and the managed-lane E2E has now been cancelled at three consecutive heads
**Location:** `seatunnel-engine-server/src/test/.../task/source/`;
`ManagedSourceRuntimeIT.java`; fork CI run `31704069417` for head `2eb0e83782`
**Problem description:** There is **no test class for
`ManagedSourceCoordinatorRuntime` (1920 lines) and none for
`ManagedSourceReaderRuntime` (1669 lines)**. The only contact with the
coordinator runtime anywhere in the test tree is
`ManagedSourceCheckpointStateTest:152-158` calling the static pure function
`reconcileNoMoreSplitsSubtasks`. Nothing drives `runOneTurn()`,
`applyBarrier()`, the restore path, or the `PREPARE_CLOSE` shutdown sequence.
`SourceSplitEnumeratorTaskTest` gained 25 lines of config plumbing and **zero**
managed-lane coverage — all four tests use the 3-arg legacy constructor.
Roughly 820 lines of new unit test cover roughly 7,800 lines of new runtime
code, and none of it covers the event loops.
**CI status, checked live at review time.** The apache-side "Build" check is
`FAILURE` (completed 2026-08-14T02:36:54Z). The real signal is the fork run
`DanielLeens/seatunnel` `31704069417` for head `2eb0e83782`: 69 jobs succeeded,
2 failed, 8 cancelled.
- `unit-test (11, windows-latest)` **failed**, but the cause is
`OptionRulesApiTest.before:73 » IllegalState Node failed to start!` — a
Hazelcast node-startup failure in a test that is not in this PR's 104-file
diff. 373 tests ran with 1 error. Not a regression from this branch.
- `doris-connector-it (11, ubuntu-latest)` **failed** on `DorisErrorIT`,
also unrelated to this diff.
- Every managed-source unit test that ran **passed**:
`ManagedSourceRuntimeConfigTest` (3), `ManagedSourceRuntimeSelectorTest` (7),
`SourceConfigTest` (1), `SourceRuntimeMetricsTest` (4),
`ManagedCoordinatorSchedulerTest` (2), `ManagedSourceCheckpointStateTest` (10),
`ManagedSourceRuntimeMetricsTest` (2), `ReaderCommandMailboxTest` (2),
`SourceAssignmentTrackerTest` (4), `SourceCommandCodecTest` (2),
`SourceSplitEnumeratorTaskTest` (4), `JobCheckpointUtilsTest` (4) — all
`Failures: 0, Errors: 0`. This confirms the protocol, codec and config layers
are healthy, which was never in question.
- **Both `engine-v2-it` legs (JDK 8 and 11) were cancelled**, as they were
at `9c2ffde8` and at `2da2feed6`. `engine-v2-it` is the only job that builds
and runs `ManagedSourceRuntimeIT`.
**Why this matters more than usual here.**
`batch_fake_to_inmemory_managed.conf` is a `job.mode = "BATCH"` FakeSource job
with `checkpoint.interval = 5000`, parallelism 2 and 5 splits. That is
precisely the shape that hits Issue 32 (silent hang on `NO_MORE_SPLITS`) and
then Issue 1 (`PREPARE_CLOSE` deadlock). **My source-level analysis predicts
`ManagedSourceRuntimeIT#testBatchJobCompletesOnManagedLane` times out at this
head** — and the one job that would confirm or refute that has never been
allowed to run to completion, at any head of this PR, for three different
infrastructure reasons.
**Best improvement:** Add direct owner-loop unit tests for both runtime
classes that do not depend on slow, flaky, unrelated jobs finishing first; and
get one `engine-v2-it` run to completion — pass or fail — on a head that also
includes the Issue 1, 31 and 32 fixes.
**Severity:** Medium, but treated as a merge gate
**Raised by another reviewer:** No
---
# 2. Code Quality Assessment
## 2.1 Coding Standards and Comment Completeness
Formatting, ASF headers, import hygiene and naming are clean throughout;
Spotless would be a no-op. Comment quality is **well above average for this
repository** — most new classes carry a substantive class-level Javadoc
explaining the invariant they maintain, not just what they are, and several
comments explain *why* (for example
`SourceAssignmentTracker.checkpointCompleted`'s explanation of why release must
happen at completion and not before, `:269-279`).
The problem is that three of the most load-bearing comments are false, and
in two cases the code immediately below them does the opposite of what they say:
- `ManagedSourceCoordinatorRuntime.java:575-583` states the restore path
must not bind ownership, then binds it ten lines later through
`assignmentTracker.restore()` (Issue 2).
- `PhysicalPlanGenerator.java:624-626` says the lane "is decided once, here,
and then persisted with the physical plan." There is no persisted physical plan
(Issue 3).
- `SourceRuntimeMetrics.java:151-157` and
`ManagedSourceRuntimeMetrics.java:313-319` describe the suffix as
"bounded-cardinality" while embedding a fresh flake id per deployment (Issue
24).
A comment that asserts an invariant the adjacent code violates is worse than
no comment, because it stops the next reader from checking. These three should
be treated as defects in their own right.
New-class Javadoc coverage is otherwise complete; new public methods and
non-trivial fields are documented. `ManagedSourceCoordinatorRuntime`'s complete
absence of logging (Issue 5) is the one place where the class falls below what
an operator will need.
## 2.2 Test Coverage and Test Stability
**Stability rating: Risk present.**
Not "Unstable": the new unit tests are deterministic, self-contained, fast,
and assert real invariants rather than happy-path round-trips.
`ManagedSourceCheckpointStateTest:115-118` and `:137-140` flip a bit in the
serialized bytes and assert `IOException` — genuine corruption-detection
coverage, and exactly the right instinct. No shared mutable statics, no
hard-coded ports, no unseeded randomness in the unit tests.
Not "Stable" either, for three independent reasons:
1. **The code that will actually break is untested** (Issue 26). Roughly
3,600 lines of event-loop, ownership, lifecycle and barrier-interleaving logic
have no direct unit test at all.
2. **`ManagedSourceRuntimeIT`'s lane assertion is order-dependent and
vacuous in one direction** (Issue 27). `server` is one container started in
`@BeforeAll` and `server.getLogs()` is cumulative for the whole class. Both
`testBatchJobCompletesOnManagedLane` (`:72-78`) and the streaming test
(`:102-104`) assert the same `MANAGED_LANE_LOG` string against that shared log,
so if the batch test runs first the streaming assertion passes without the
streaming job ever entering the managed lane. The class's own Javadoc
(`:42-45`) says that without a real lane assertion the suite "would be worse
than no coverage at all" — which is the state it is in for the second test.
Both `jobFuture.get()` (`:110`) and `restoreFuture.get()` (`:134`) are untimed,
so a hung job blocks the CI worker rather than failing it — directly relevant
given Issues 1 and 32.
3. **`SourceSplitEnumeratorTaskTest`, now load-bearing for this feature's
config plumbing, carries pre-existing flake sources**: a `Thread.sleep(1000)`
plus a strict `<` comparison on two `System.currentTimeMillis()` stamps
(`:114`, `:129`) that fails if both land in the same millisecond; an unbounded
busy-wait `while (openTime.get() == 0 || ...) { enumeratorTask.call(); }` with
no timeout (`:125-127`); and a negative timing assertion
`assertFalse(latch.await(200, MILLISECONDS))` (`:242`) that a GC pause inverts.
Suggested minimum before merge: a `ManagedSourceCoordinatorRuntimeTest`
driving `stateProcess()` through `RUNNING -> PREPARE_CLOSE -> CLOSED` with a
queued barrier (Issue 1); a restore test asserting the tracker is usable from
the owner loop after `restoreRuntimeState` on a different thread (Issue 2); a
`ManagedSourceReaderRuntimeTest` asserting a bounded reader reaches
`END_OF_INPUT` after `NOTHING_AVAILABLE` followed by `NO_MORE_SPLITS` (Issue
32); and a dispatch test asserting a `!ready` reader defers rather than throws
(Issue 31). Each is a few dozen lines and each pins a current High.
## 2.3 Documentation Updates
`docs/en` and `docs/zh` are both present, structurally parallel, and
correctly wired into `docs/sidebars.js:284`. The configuration table in
`managed-source-runtime.md:100-126` was checked field-by-field against
`ManagedSourceRuntimeConfig.java:42-70` — **all 24 defaults match exactly**,
which is genuinely good discipline.
Four mismatches, in severity order:
1. **The lane-persistence claim is wrong** (`en:82-83`, `en:214-215`,
`zh:74`, `zh:201`) — Issue 3. This is the most consequential, because the
rollback procedure built on it will brick production jobs.
2. **The metric name format is wrong** (`realtime-observability.md:311` en
and zh: "All names above use the same `#{sourceId}` suffix"). Code emits
`#{sourceId}#attempt-{executionId}`. A dashboard query written from this doc
matches nothing. The same omission applies to the metric list at
`managed-source-runtime.md:170-180`.
3. **"Managed metrics use a stable Source action suffix"** (`en:165`,
`zh:154`) is the opposite of what the code does — Issue 24. The adjacent
sentence about command ids, split ids and table names never being metric
dimensions *is* accurate.
4. **No `incompatible-changes.md` entry**, and — more importantly — no
documentation anywhere of the three operator-facing hazards: config flip makes
managed jobs unrestorable, connector-jar upgrade invalidates every managed
checkpoint via the digest, and an allowlist typo hard-fails every job using
that connector at submit *and* at restore.
# 3. Architectural Soundness
## 3.1 Elegance of the Solution
**This is a long-term solution, correctly shaped, incompletely executed.**
It is not a workaround and not a point fix. Moving Source concurrency ownership
from every connector author into the engine is the right call, and the specific
mechanisms chosen are the ones a mature system would choose: an explicit
versioned command envelope rather than ad-hoc messages; CRC-checksummed,
magic-prefixed, length-bounded, explicitly non-Java serializers rather than
`ObjectOutputStream`; an engine-owned assignment ledger with an explicit state
machine rather than implicit ownership; attempt fencing via a monotonic
deployment id; bounded mailboxes with reserved control capacity so control
traffic cannot be starved by data traffic; a scheduler with explicit overlap
and failure policies rather than raw threads.
The steady-state protocol is well developed. What is missing is the
transition machinery, and it is missing consistently: shutdown (Issue 1),
restore (Issue 2), cancel (Issue 8), coordinator epoch change (Issue 21),
reader failover (Issue 31) and bounded completion (Issue 32) are all broken or
unimplemented. That is a coherent gap rather than scattered sloppiness, which
is encouraging — it suggests one focused pass rather than a rewrite.
## 3.2 Maintainability
Two 1,700-to-1,900-line classes with hand-rolled event loops, mailboxes,
retry and deadline machines and thread-ownership assertions, carrying **zero
direct unit tests** and, in the coordinator's case, **zero log statements**.
That combination is the main maintainability concern, ahead of any individual
bug.
The `checkOwner()` lazy first-caller-wins pattern is the main structural
concern and is the direct cause of two Highs (Issues 2 and 9) plus part of
Issue 29. Replacing it with an explicit `attachToOwnerLoop()` called once from
the first `runOneTurn()` — with a clear "coordinator loop has not started"
error for any earlier caller — would make both bug classes impossible by
construction rather than by convention, and would make the
`ThreadShareMode.ALL` incompatibility a startup-time rejection instead of a
turn-two crash.
## 3.3 Extensibility
Good. `runtimeProtocolVersion`, `connectorStateVersion` and
`capabilityDigest` give three independent evolution axes; wire codes are
explicit enums with integer codes; serializers are versioned with magic and
checksum. Two gaps: a capability-flag change is currently an unrecoverable
restore break with no migration story (Issue 5b), and
`runtime-protocol-version` is never bounded against the engine's actual
supported version, so a v2-labelled deployment would silently run v1 wire code.
## 3.4 Historical-Version Compatibility
Covered in 1.2. The SPI surface is fully compatible and the serializer IDs
are append-only — both done properly. The restore and deployment paths are not:
Issue 3 (lane re-derivation), Issue 6 (`FakeSourceSplit` identity, legacy lane,
ungated), Issue 25 (`SourceConfig` serialization), and the undocumented digest
gate. None of it appears in `incompatible-changes.md`.
# 4. Issue Summary
| No. | Issue | Location | Sev | Status |
|---|---|---|---|---|
| 1 | Enumerator deadlocks in `PREPARE_CLOSE`; control events never drained
| `SourceSplitEnumeratorTask.java:466` | High | Open, re-derived |
| 2 | Restore binds tracker owner to the restore thread; all managed
restores fail | `ManagedSourceCoordinatorRuntime.java:593`,
`SourceAssignmentTracker.java:345` | High | Open, re-derived |
| 3 | Lane re-derived from live config on every init; docs claim the
opposite; rollback bricks jobs | `PhysicalPlanGenerator.java:170,612`,
`JobMaster.java:290` | High | Open, re-derived |
| 4 | Iceberg declares `supportsAsyncEnumerator` but the BOUNDED path blocks
| `IcebergSource.java:136,149,166` | High | Open, re-derived |
| 5 | Managed ops collapse all exceptions to `STALE_TARGET`; coordinator has
zero logging | `ManagedSourceCommandOperation.java:70` +2 | High | Open, scope
corrected |
| 6 | `FakeSourceSplit` identity widened -> enumerator dedup set misses ->
**duplicate data on upgrade restore**, legacy lane, ungated |
`FakeSourceSplit.java:22-43`, `FakeSourceSplitEnumerator.java:135` | High |
Open, mechanism now concrete |
| 7 | Watchdog `interrupt()` outside the lock; no re-validation; flag never
cleared | `ManagedSourceReaderRuntime.java:601` | High | Open, re-derived |
| 8 | CANCEL cannot preempt a pending barrier; clean cancel reported FAILED
| `ManagedSourceReaderRuntime.java:1487,1546` | High | Open, mechanism
corrected |
| 9 | Thread-identity guards break under `ThreadShareMode.ALL` |
`TaskExecutionService.java:654,1387` +4 | High | Open, re-derived |
| **31** | **Reader mid-replacement turns `assignSplit` into a whole-job
failure** | `ManagedSourceCoordinatorRuntime.java:1500`,
`SeaTunnelSplitEnumeratorContext.java:70` | **High** | **NEW — carryover miss**
|
| **32** | **`NO_MORE_SPLITS` never re-opens the poll gate; bounded reader
hangs forever** | `ManagedSourceReaderRuntime.java:702` | **High** | **NEW —
carryover miss** |
| 10 | Epoch/attempt/`closed` read cross-thread without `volatile` |
`ManagedSourceReaderRuntime.java:114,136,256` | Med | Open |
| 11 | Deferred split request dropped when the reader is not ready |
`ManagedSourceCoordinatorRuntime.java:1410` | Med | Open |
| 12 | `enumerator.run()` inline on the drain thread blocks all barriers |
`SourceSplitEnumeratorTask.java:448` | Med | Open |
| 13 | Kafka managed discovery kills the job on transient errors; re-arm
sits past the `throw` with no `finally` |
`KafkaSourceSplitEnumerator.java:584-601` | **High** | Open, severity raised |
| 14 | Kafka `end_offsets_timestamp` — refuted as stated; residual is
dynamic-discovery rounds only | `KafkaSourceSplitEnumerator.java:749` | Low |
Corrected |
| 15 | `restoreMetadata()` lacks owner check and idempotence guard |
`ManagedSourceReaderRuntime.java:194` | Med | Open |
| 16 | Three non-failure-atomic release paths leak the node-wide budget |
`ManagedSourceReaderRuntime.java:1246,506`, `ReaderCommandMailbox.java:278` |
Med-High | Open, relocated |
| 17 | `containsAll` reconciliation replays partially finished chunks on
failover | `SourceAssignmentTracker.java:257` | Med | Open |
| 18 | Transport retry maps leak on reader replacement |
`ManagedSourceCoordinatorRuntime.java:1172` | Med | Open |
| 19 | Managed ack failure never logged; can be lost after close;
coordinator variant worse | `SourceSeaTunnelTask.java:198`,
`SourceFlowLifeCycle.java:846` | Med | Open, scope corrected |
| 20 | Per-turn metrics update double-locks and scans the mailbox |
`ManagedSourceReaderRuntime.java:392` | Med | Open |
| 21 | Epoch rebase unimplemented; `resetForEpoch`/`awaitSignal` dead;
watermark never read | `ManagedSourceReaderRuntime.java:741` | Med | Open |
| 22 | Enumerator client close vs in-flight discovery — engine closes the
scheduler first; small residual window | `KafkaSourceSplitEnumerator.java:285`
| Low | Corrected |
| 23 | Concrete `pollNextManaged` makes the `instanceof ManagedSourceReader`
guard vacuous for the whole `SourceReaderBase`/CDC family |
`SourceReaderBase.java:61,133`; `SourceFlowLifeCycle.java:222` | Med |
Corrected, relocated |
| 24 | Metric names embed the task attempt; contradicts docs and own
Javadoc; hits legacy lane | `SourceRuntimeMetrics.java:158,207` | Med | Open |
| 25 | `SourceConfig` +5 fields, no `serialVersionUID`; no
`incompatible-changes.md` entry | `SourceConfig.java:26-30` | Med | Open |
| 26 | Owner-loop classes untested; managed-lane E2E cancelled at three
consecutive heads | new UT dir, `ManagedSourceRuntimeIT.java` | Med (gate) |
Open, CI re-checked |
| 27 | E2E lane assertion order-dependent on an accumulating shared
container log | `ManagedSourceRuntimeIT.java:102` | Med | Open |
| 28 | Poll-status drain path leaves `prepareClose` false, so flush signals
continue | `ManagedSourceReaderRuntime.java:441` | Low | Open, largely refuted |
| 29 | `requestSplit()`/`signalNoMoreElement()` throw before the epoch
handshake | `ManagedSourceReaderRuntime.java:1075` | Med | Open |
| 30 | Grouped low-severity items (dead code, wire hygiene, error messages,
config, style) | see Issue 30 | Low | Open |
| **33** | **`checkpointAssignmentProofs`/`uncheckpointedAssignmentSplitIds`
unbounded** | `ManagedSourceReaderRuntime.java:109` | **Med** | **NEW** |
| **34** | **`subscribeToAvailability()` accumulates handlers on a reused
future** | `ManagedSourceReaderRuntime.java:641` | **Med** | **NEW** |
| **35** | **Scheduler timeout `cancel(true)` leaks the interrupt flag onto
shared pools** | `ManagedCoordinatorScheduler.java:352,383` | **Med** | **NEW**
|
| **36** | **Kafka split-back handler calls `assignSplit()` eagerly, racing
reader readiness; silent partition loss variant** |
`KafkaSourceSplitEnumerator.java:635,510` | **High** | **NEW** |
| **37** | **`SourceReaderBase.isAvailable()` short-circuits on
`noMoreSplitsAssignment`, degrading to a fixed 100 Hz poll** |
`SourceReaderBase.java:172` | **Med** | **NEW** |
| **38** | **`FakeSourceReader.pollNextManaged` throws on a short read ->
permanent crash loop after a config edit plus restore** |
`FakeSourceReader.java:140-153` | **Med** | **NEW** |
**Refuted or substantially narrowed this round:** codec unchecked length
reads (not present — every read is bounded by `in.available()` before
allocation); unbounded silent retry loop on `STALE_TARGET` (deadline-bounded in
both directions); swallowed checkpoint ack on the legacy lane (legacy `.join()`
byte-for-byte preserved); `ManagedSourceMemoryBudget` not failure-atomic (it
is); managed lane never sets `prepareClose` (it does); Kafka
`end_offsets_timestamp` outside TIMESTAMP mode (config only ever sets it in
TIMESTAMP mode); Kafka client close vs in-flight discovery (engine closes the
scheduler first); CDC readers broken by inherited `pollNextManaged` (CDC cannot
be selected into the lane at all); silent cross-lane checkpoint restore
(fail-closed in all four directions); config-table drift (all 24 defaults
match); `SourceReaderBase` exhausted-batch infinite loop, `FetchTask` lost
wakeup, `SplitFetcher.wakeUp()` deadlock, and dropping the checkpoint lock in
`pollNextManaged` (
all verified correct by design).
**Reach classification, which should drive the verdict.** Exactly **one**
finding reaches production unconditionally: Issue 6, on the legacy lane, with
the feature flag off. Everything else is opt-in behind `enabled: true` plus an
explicit allowlist entry, or latent until a connector is onboarded.
# 5. Merge Recommendation
### Conclusion: Not recommended for merge
Two process notes before the substance. First, **I am the author of this
PR.** GitHub does not permit an author to approve or request changes on their
own pull request, so my account can only leave a comment-type review regardless
of the conclusion below. Please read this with that bias disclosed. The final
call has to come from an independent maintainer with write access. Second,
**this PR is still a Draft** (`isDraft: true`), and `mergeStateStatus` is
`BLOCKED` with `reviewDecision: REVIEW_REQUIRED` — both expected for a draft
awaiting a write-capable reviewer, not new problems. `mergeable` is
`MERGEABLE`; there are no conflicts with `dev`.
**There is no existing maintainer APPROVED or CHANGES_REQUESTED review on
this PR to respond to.** The review history is six comment-type reviews from me
and one comment-type review from @nzw921rx carrying a single inline comment. So
no merge intent from an independent maintainer has been expressed, and nothing
in this review contradicts a maintainer's stated position.
The technical picture did not change with this commit, because the commit is
a pure `dev` sync. What changed is my own confidence and, unfortunately, the
issue count: a cold re-read surfaced **two new High-severity findings I missed
in five prior rounds** (Issues 31 and 32), **three new Medium findings**
(33-35), and **five places where my own earlier reports were wrong or
overstated** (Issues 5, 8, 16, 19, 28), which I have corrected above and
apologised for. Issue 32 in particular means a bounded managed job hangs
silently *before* it can even reach the `PREPARE_CLOSE` deadlock I have been
reporting since round one.
### 1. Blockers — must be fixed, sorted by severity
1. **Issue 32 (High, NEW)** — `NO_MORE_SPLITS` never re-opens the poll gate.
A bounded managed reader hangs silently forever. One-line fix; highest
severity-to-effort ratio in this list.
2. **Issue 1 (High)** — the managed enumerator deadlocks in `PREPARE_CLOSE`.
No managed bounded job can complete and no savepoint can finish.
3. **Issue 2 (High)** — coordinator restore poisons the assignment tracker's
owner thread. No managed job recovers from any checkpoint or savepoint,
non-deterministically.
4. **Issue 31 (High, NEW)** — a routine reader failover turns
enumerator-initiated `assignSplit` into a whole-job failure. The deferral
mechanism already exists for split requests; it just needs extending.
5. **Issue 8 (High)** — CANCEL cannot preempt a pending barrier, so every
cancel racing an in-flight checkpoint reports FAILED instead of CANCELED, and a
post-cancel barrier still snapshots a cancelled reader.
6. **Issue 3 (High)** — lane selection is re-derived from live cluster
config on every init. The documented rollback procedure permanently bricks
running managed jobs. Either persist the selection or delete the four false
claims and document the real hazard.
7. **Issue 9 (High)** — the four thread-identity guards are incompatible
with the supported `ThreadShareMode.ALL`. Reject the combination at selection
time at minimum.
8. **Issue 7 (High)** — the poll watchdog can deliver an interrupt to
unrelated work and never clears the flag, so cleanup itself fails and connector
resources leak.
9. **Issue 4 (High)** — Iceberg's capability declaration does not match its
bounded enumerator.
10. **Issue 5 (High)** — all three managed operations erase the root cause,
and `ManagedSourceCoordinatorRuntime` has zero logging in 1,920 lines. This one
is not a correctness bug, but shipping an engine feature with no diagnostics is
not something I would sign off on.
11. **Issue 36 (High, NEW)** — Kafka's split-back handler calls
`assignSplit()` eagerly, which both races reader readiness into an intermittent
job failure and, in the all-readers-unregistered case, silently drops
partitions from every future discovery snapshot. Deleting one line fixes it.
12. **Issue 13 (High, severity raised)** — Kafka's managed discovery kills
the job on a transient broker error where legacy retried forever, and the
re-arm call sits past the `throw` with no `finally`, so the obvious fix
silently disables discovery instead. Both halves must be fixed together.
13. **Issue 6 (High, and the only unconditional defect in the PR)** —
`FakeSourceSplit`'s widened identity makes the enumerator's dedup set miss on
any restore across the upgrade boundary, re-emitting already-consumed splits,
on the **legacy** lane with the feature flag off. One-annotation fix, plus a
test that actually serializes.
14. **Issue 26 (Medium, treated as a gate)** — `engine-v2-it` has now been
cancelled at three consecutive heads, so there is still **zero CI evidence at
any head of this PR** that the managed lane's shutdown, restore or cancel paths
behave as designed. I would not consider this gate satisfied until one
`ManagedSourceRuntimeIT` run completes — pass or fail — on a head that also
includes the Issue 1, 31 and 32 fixes.
**If only one thing is fixed before anything else, make it Issue 6** — it is
the sole defect that affects users who never enable this feature.
### 2. Recommended fixes — non-blocking
Issues 10-25, 27-30, 33-35 and 37-38 should be addressed before the feature
graduates from experimental. Issue 23 deserves early attention despite its
Medium rating: making `SourceReaderBase.pollNextManaged` abstract is a small
change that restores the engine's `instanceof ManagedSourceReader` guard to
actually meaning something, and it is the trap the next connector onboarding
will otherwise fall into. Several are latent correctness or leak issues that
will be far harder to diagnose once the lane is enabled anywhere: Issue 10
(missing `volatile` producing spurious `STALE_TARGET` right after the
handshake), Issue 16 (permanent node-wide budget leak), Issue 33 (unbounded
growth exactly when checkpointing is already stuck), Issue 35 (interrupt
leaking across jobs on shared pools), Issues 11, 17, 18 (silent split-handling
anomalies). Issue 30 is cleanup, though the `validateRestoredSelection` error
message is worth fixing early — it is the message an operator is most likely to
hit
and the least useful one in the PR.
### 3. Response to @nzw921rx
On the connector allowlist
(`docs/zh/engines/zeta/managed-source-runtime.md:31`): thank you for raising
this, and I want to engage with it properly rather than just restate my earlier
position, because part of your argument is now better supported by evidence
than my counter-argument was.
**Where you are right, with new evidence.** You argued that this should be a
feature switch and that an unsupported connector should fail fast rather than
be governed by a separate list. The code already does the fail-fast half of
what you describe: `ManagedSourceRuntimeSelector.java:49-55` hard-throws when
an allowlisted connector declares no managed capability. So the allowlist is
not preventing silent weakening — the capability check is. Your point that its
effect "will gradually weaken" as connectors adopt the capability is also
correct: once a connector declares a capability, the allowlist is the only
thing standing between it and the managed lane, and maintaining two switches
for one decision is real long-term cost.
**Where I would push back, with counter-evidence.** The allowlist's value
today is not semantic, it is blast-radius control during rollout, and this
round turned up a concrete reason that matters more than I previously argued.
Because of Issue 3, the lane is recomputed from live cluster config on
**every** `JobMaster#init`, including restore and master failover. With a
per-connector allowlist, an operator enabling the feature affects only the
connectors they named. Without it — a single global `enabled: true` — flipping
the switch changes the lane for *every* capability-declaring connector in the
cluster at once, and flipping it back makes every job that took a managed
checkpoint in the interim permanently unrestorable. That is a cluster-wide,
unrecoverable outage from one config edit.
There is also a hazard in the current allowlist that argues for tightening
it rather than removing it: because the check throws rather than falling back,
adding a non-capable connector name — a plausible typo, or an over-broad
rollout entry like `jdbc` — hard-fails **every** job using that connector, both
new submissions and restores of already-running jobs. That blast radius is
undocumented.
**Concretely, what I think should happen.** Keep the allowlist for phase 0,
but (a) fix Issue 3 so the selection travels with the job, which is what makes
any switch safe to flip; (b) change the allowlisted-but-incapable case to
`WARN` and fall back to legacy, since the operator's intent is unambiguously
"use managed if capable" — that directly implements the fail-safe behaviour you
are asking for without the outage risk; and (c) document the removal path, so
the allowlist can be retired once Issue 3 is fixed and adoption is broad. If a
maintainer prefers your simpler model, I would accept it as long as (a) lands
first — without persisted selection I think a single global switch is genuinely
dangerous.
### Overall assessment
The right architecture, roughly 80% built, with the transition machinery
still the missing 20% — and this round that gap got larger, not smaller,
because two of the three ways a bounded managed job fails to finish were
previously unreported. The steady-state protocol, the fail-closed compatibility
gating, the versioned checksummed serializers and the append-only serializer
IDs remain genuinely good work, and they are why the remaining problems are
localized rather than systemic. I want to be clear that my conclusion is about
executable correctness on the primary transition paths, not about the design,
which I still think is the right one for this project.
**Option A (recommended): finish phase 0 as scoped.** Fix the twelve
blockers, add the four small owner-loop unit tests named in section 2.2, and
get one `engine-v2-it` run to completion. That is a focused, bounded amount of
work — most of the blockers are localized, and Issues 32, 6 and 4 are close to
one-liners. Then this is mergeable as an experimental, disabled-by-default
feature.
**Option B: narrow phase 0 to the coordinator lane only.** Ship
`MANAGED_COORDINATOR` and defer `MANAGED_READER_AND_COORDINATOR` to a
follow-up. This removes `ManagedSourceReaderRuntime` — 1,669 lines carrying
Issues 7, 8, 10, 15, 20, 21, 29, 32, 33, 34 — plus Issues 23, 37 and 38 from
the merge surface entirely, cutting the blocker list roughly in half while
still delivering the engine-owned enumerator, the scheduler and the assignment
ledger, which are the parts most connectors would benefit from first. Issues 1,
2, 3, 4, 5, 6, 9, 13, 26, 31 and 36 would still need fixing. Note that Issue 6
must be fixed under **either** option, since it does not depend on the lane at
all. If the goal is to get value into `dev` sooner and reduce review surface, I
think Option B is genuinely worth considering.
I do not think there is a materially better Option C for the overall design.
The one structural change I would still push for in either option: replace the
lazy `checkOwner()` first-caller-wins bind with an explicit
`attachToOwnerLoop()` invoked once from the first `runOneTurn()`. That closes
Issue 2 by construction, turns Issue 9 into a clear startup-time rejection
instead of a turn-two crash, and removes a whole class of future bug.
Thank you again for the care that went into the documentation, the
compatibility gating and the serializers — the fail-closed posture, the
explicit capability handshake and the checksummed versioned wire format are the
right instincts, and the 24-of-24 config-table match is the kind of discipline
that is rare in this repo. The thing I would most like to see before the next
round is still the same: one completed `ManagedSourceRuntimeIT` run. My
analysis now predicts the batch test hangs for two independent reasons; a real
run is the single piece of evidence most likely to move my assessment in either
direction.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
