ericyuan915 opened a new issue, #19375:
URL: https://github.com/apache/hudi/issues/19375
### Bug Description
**What happened:**
A bounded read via the new FLIP-27 `HoodieSource` (Source V2,
`read.source-v2.enabled=true`), when restored from a checkpoint or savepoint,
reuses the split set persisted in the enumerator state and never re-enumerates.
If the job is resumed with a **changed read scope** — a different
`read.start-commit` / `read.end-commit`, or a different partition selection —
the change is silently ignored and the job re-reads the **old** scope captured
in the checkpoint. There is no exception and no warning; the job runs to
completion and simply produces data for the wrong scope.
**What you expected:**
Resuming a bounded read whose configured scope no longer matches the
checkpoint should not silently read stale data. At minimum the mismatch should
be detected and surfaced (fail fast with an actionable message, or explicitly
handle the new scope) rather than producing incorrect results with no signal.
**Steps to reproduce:**
1. Enable Source V2 (`read.source-v2.enabled=true`) and configure a
**bounded** read of a Hudi table — e.g. a snapshot/incremental query with
`read.start-commit=<A>` and `read.end-commit=<B>` (or a static
partition selection). Run in **streaming execution mode with checkpointing
enabled** and externalized/retained checkpoints (`BATCH` mode does not
checkpoint, so streaming mode is required to have a
checkpoint to resume from).
2. Let the job take at least one checkpoint, then stop/cancel it.
3. Resume from that checkpoint/savepoint (`flink run -s <path> ...`) but
**change the scope** — e.g. widen `read.end-commit=<C>` or add/remove
partitions.
4. Observe: the job reads the **original** `[A, B]` scope from the
restored splits, not the newly-configured one. Nothing is logged about the
discrepancy.
**Root cause:**
`HoodieSource#createEnumerator` re-derives splits via
`createBatchHoodieSplits()` **only when the restored enumerator state is
null**. On restore (`state != null`) it takes the pending splits straight from
`HoodieSplitEnumeratorState#getPendingSplitStates()` and never re-enumerates,
and there is no comparison between the restored scope and the
currently-configured scope. `HoodieSplitEnumeratorState` also carries no
representation of the read scope, so a mismatch cannot be detected. A bounded
read's split set is therefore frozen at first enumeration;
`allowNonRestoredState` does not help (it only governs operator-state mapping,
not the enumerator's split set).
> Confirmed on `master`: in `createEnumerator`, `createBatchHoodieSplits()`
is guarded by `enumeratorState == null`; the restore branch maps
`getPendingSplitStates()` to splits with no scope validation.
**Proposed fix:**
Persist a compact **scope token** in `HoodieSplitEnumeratorState` — table
path, table type, query type, start/end commit bounds, and the pruned partition
set (deliberately **excluding** projection / `requiredColumns`, which change
what is read per-row but not the split set). Bump the enumerator-state
serializer version and version-gate the read so pre-existing checkpoints (no
token) are allowed through. On restore of a **bounded** read, compare the
checkpointed token to the configured token; on mismatch, fail the job with an
actionable message (*"scope changed since checkpoint — start a fresh backfill /
use a new checkpoint dir"*). Streaming reads are left unguarded (they
legitimately resume-and-continue).
Two implementation notes that matter (happy to discuss the preferred
direction before sending a PR):
1. **The failure must be raised from the enumerator event loop
(`SplitEnumerator#start()`), not thrown from `createEnumerator` /
`restoreEnumerator`.** A throw during restore executes inside
`SourceCoordinator`'s asynchronous reset
(`RecreateOnResetOperatorCoordinator`), where the resulting `failJob` does
**not** terminate the job — it comes up `RUNNING` with no enumerator, zero
throughput, and empty checkpoints (a "zombie"). Raising it from `start()` goes
through `SourceCoordinator#runInEventLoop` → `context.failJob`, which fails the
job for real; wrapping it in `SuppressRestartsException` keeps
the restart strategy from looping on the same mismatch. (Verified against
Flink `release-1.18` and `release-1.20`.)
2. **Re-enumerating in place is not a correct fix by itself.** FLIP-27
reader state is checkpointed independently of the enumerator, so after
discarding enumerator state and re-enumerating, readers would still resume and
finish their **old** in-flight splits → the job reads `(new enumerated scope) ∪
(old in-flight reader splits)`. A correct scope change must reset **both**
enumerator and reader state, which is equivalent to a fresh start. Hence
fail-fast (force a clean restart) rather than silent re-enumerate. An opt-in
"re-enumerate + reset reader state" or a warn-only mode are alternatives worth
considering.
### Environment
- **Hudi version:** `master` (1.3.0-SNAPSHOT). Source V2
(`org.apache.hudi.source.HoodieSource`) is recent — added starting #17989
(2026-01-23); not in 1.2.0 or earlier.
- **Query engine:** Flink — FLIP-27 Source V2
(`read.source-v2.enabled=true`), wired into `HoodieTableSource` since #18022.
Affects bounded snapshot and incremental reads.
- **Relevant configs:** `read.source-v2.enabled=true`; a bounded scope
(`read.start-commit` / `read.end-commit` or a static partition selection);
streaming execution mode with checkpointing + externalized/retained checkpoints
(or savepoint-based resume).
### Logs and Stack Trace
None — the failure mode is **silent**: no exception and no warning are
produced; the job succeeds while reading the checkpoint's old scope. That
silence is the core of the report. (With the proposed guard, the mismatch
instead surfaces as a job failure carrying the checkpoint-vs-configured scope
diff and remediation steps.)
--
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]
