fix(flink): fail fast when a bounded HoodieSource resumes a checkpoin… - #19376
fix(flink): fail fast when a bounded HoodieSource resumes a checkpoin…#19376ericyuan915 wants to merge 1 commit into
Conversation
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR adds a scope guard for bounded (batch) FLIP-27 HoodieSource reads: it captures a scope token (path/table type/query type/start+end commit/partitions) into the checkpointed enumerator state (serializer bumped v1→v2), and on restore fails the job from start() (via a deferred SuppressRestartsException) if the configured scope changed. The serialization versioning, backward-compat handling of pre-guard checkpoints, and the deferred-failure mechanism all look carefully thought through. One gap worth double-checking on the partition dimension of the token is noted inline. Please take a look at the inline comment, and this should be ready for a Hudi committer or PMC member to take it from here. One readability nit below — the inline comment in createEnumerator largely duplicates the Javadoc on computeBoundedScopeFailure.
| * for pruners whose selection is not a fixed partition list (e.g. dynamic / column-stats | ||
| * pruning); {@link StaticPartitionPruner} overrides it with the concrete partition set. | ||
| */ | ||
| default String scopeToken() { |
There was a problem hiding this comment.
🤖 Only StaticPartitionPruner overrides scopeToken() with the concrete partition set — but in the real HoodieTableSource path partition filters always become partitionEvaluators → DynamicPartitionPruner (nothing in main ever calls .candidatePartitions(...), so StaticPartitionPruner is effectively test-only). That means the partitions= component of the token is a constant class name in production, so changing the partition WHERE clause won't change the token and the guard won't catch it — even though "partition selection" is one of the listed target scenarios. The tests exercise only the static path, so this gap isn't caught. Could you confirm whether dynamic pruners were meant to contribute a meaningful token (e.g. from the evaluators/probe)? @danny0405
| // legitimately resume-and-continue, so they are not guarded (token stays empty). | ||
| Option<String> currentScopeToken = | ||
| streaming ? Option.empty() : Option.of(computeBoundedScopeToken(scanContext)); | ||
| // Deferred bounded-scope failure. A bounded read restored from a checkpoint whose scope changed |
There was a problem hiding this comment.
🤖 nit: this 7-line block restates almost word-for-word the Javadoc already written on computeBoundedScopeFailure. Could you trim it to a one-liner like // Deferred to avoid zombie job — see computeBoundedScopeFailure. and let the Javadoc be the single source of truth?
| * for pruners whose selection is not a fixed partition list (e.g. dynamic / column-stats | ||
| * pruning); {@link StaticPartitionPruner} overrides it with the concrete partition set. | ||
| */ | ||
| default String scopeToken() { |
There was a problem hiding this comment.
Production partition filters use DynamicPartitionPruner, whose scopeToken() falls back to the class name. Since candidatePartitions is only used in tests, changing the partition predicate produces the same token and the restored stale splits are still accepted. Could the dynamic token include a canonical representation of the predicate and its bound values?
There was a problem hiding this comment.
You're right. Rather than build a canonical predicate form, I've dropped partitions from the guard entirely and reverted PartitionPruners to master, per danny's high-level point that these options are the user's responsibility. The guard is now commit-range only.
| String partitions = scanContext.getPartitionPruner() == null | ||
| ? "none" | ||
| : scanContext.getPartitionPruner().scopeToken(); | ||
| return String.join( |
There was a problem hiding this comment.
The scope token omits settings that affect bounded incremental split enumeration, including CDC mode, read.cdc.from.changelog, and the skip-compaction/clustering/insert-overwrite options. Restoring after changing these settings would still accept the old splits. Could we include all split-generation inputs in the token?
There was a problem hiding this comment.
After chatted with Danny, we decide to only validate the date ranges, will skip the rest.
| // legitimately resume-and-continue, so they are not guarded (token stays empty). | ||
| Option<String> currentScopeToken = | ||
| streaming ? Option.empty() : Option.of(computeBoundedScopeToken(scanContext)); | ||
| // Deferred bounded-scope failure. A bounded read restored from a checkpoint whose scope changed |
There was a problem hiding this comment.
Is deferring this failure to start() actually necessary? In Flink 1.18/1.20, exceptions from restoreEnumerator() propagate through resetAndStart() to cleanAndFailJob(), which calls context.failJob() and triggers a global failover.
There was a problem hiding this comment.
You're right about resetAndStart — it does catch and call cleanAndFailJob. The break is one level down, and only on the initial savepoint restore, which is the case this guard exists for. OperatorCoordinatorHolder#resetToCheckpoint says in its own comment that the first call happens during ExecutionGraph construction, before lazyInitialize supplies the scheduler executor; LazyInitializedCoordinatorContext#failJob opens with checkInitialized(), which then throws IllegalStateException inside the closingFuture.whenComplete(...) callback whose future is discarded. The throw also escapes before processPendingCalls(), so hasCaughtUp stays false and the later start() only queues — the enumerator never runs. I hit exactly this internally before moving the failure to start(): job RUNNING, zero throughput. On a mid-run failover your description is accurate; it's the initial restore that differs.
| // resuming would silently read the checkpoint's OLD scope. Capture the current scope here so it is | ||
| // checkpointed (snapshotState) and can be compared on the next restore. Streaming reads | ||
| // legitimately resume-and-continue, so they are not guarded (token stays empty). | ||
| Option<String> currentScopeToken = |
There was a problem hiding this comment.
Why is scope validation disabled entirely for streaming restores? If a partition predicate is narrowed from partition IN ('p1', 'p2') to partition = 'p1', the checkpointed in-flight/pending splits for p2 are still restored, while only newly discovered splits use the new predicate. Streaming progress should resume from checkpoint, but split-selection settings such as partition predicates may still require compatibility validation.
There was a problem hiding this comment.
For the commit range specifically, streaming must stay exempt: IncrementalInputSplits:266 resolves startCompletionTime as issuedOffset != null ? issuedOffset : conf.get(READ_START_COMMIT), so a checkpointed offset deliberately supersedes read.start-commit and validating it would fail every legitimate streaming resume. The partition case is real, but it isn't fixable by config comparison — the honest fix is re-filtering restored pending splits through the current pruner on restore. Can follow up in another PR.
danny0405
left a comment
danny0405
left a comment
There was a problem hiding this comment.
I found two additional correctness issues beyond the existing scope-coverage threads; details are inline.
| @Internal | ||
| public class HoodieEnumeratorStateSerializer implements SimpleVersionedSerializer<HoodieSplitEnumeratorState> { | ||
| private static final int VERSION = 1; | ||
| private static final int VERSION = 2; |
There was a problem hiding this comment.
Bumping the outer state serializer to v2 also makes the unchanged call below use splitSerializer.deserialize(2, splitBytes), even though these bytes are still produced by HoodieSourceSplitSerializer v1. The nested deserializer currently ignores its version argument, so the tests pass, but this silently couples the two independent formats and will select the wrong split format as soon as that deserializer becomes version-aware. Could we record the nested serializer version with each payload (or explicitly map outer v1/v2 to split v1) instead of forwarding the outer version?
There was a problem hiding this comment.
Good catch — done. v2 payloads now lead with splitSerializer.getVersion() and that is what's passed to splitSerializer.deserialize(...); v1 payloads map to LEGACY_SPLIT_SERIALIZER_VERSION = 1. Also added a version > VERSION guard.
| public String scopeToken() { | ||
| // Sorted so the token is order-independent: the same partition set always yields the same | ||
| // token regardless of insertion order. | ||
| return "static(" + this.partitions.stream().sorted().collect(Collectors.joining(",")) + ")"; |
There was a problem hiding this comment.
This encoding is not collision-free for valid partition names: { "a,b", "c" } and { "a", "b,c" } both produce static(a,b,c). A restore that changes between those sets would therefore be accepted with stale splits, violating the method's stated requirement that every selection change alter the token. Could we use an unambiguous encoding (for example, length-prefix each sorted value) or hash a canonical serialization?
There was a problem hiding this comment.
Correct. Resolved by removal — PartitionPruners is back to master and the state now holds two typed fields instead of a joined string, so there's no encoding left to collide.
|
Have some high-level questions:
|
|
@danny0405 Thanks for review!
|
just like streaming source, it is possible we snapshot the consumption offset in the snapshot ans aways start from there, so that there is no need to throw and the job can continue from its last failoure. probably just log some warnings instead of throwing? |
…t with a changed commit range A bounded FLIP-27 HoodieSource read enumerates its splits once at job start. On a checkpoint or savepoint restore it reuses that persisted split set and never re-enumerates, so resuming after read.start-commit / read.end-commit were edited silently reads the checkpoint's old range with no error and no warning. Record the configured commit range in HoodieSplitEnumeratorState and compare it on restore of a bounded read, failing fast with an actionable message when it differs. An unconfigured bound is recorded as the empty string so that "recorded but unset" stays distinguishable from "not recorded at all"; a state carrying neither bound cannot be verified and is allowed through with a warning, keeping existing checkpoints restorable. Bounds are compared after normalization, so a change of case in the `earliest` sentinel or surrounding whitespace does not fail a job that is in fact reading the same data. The failure is raised from the enumerator's start() rather than at restore time. On the initial restore from a savepoint, OperatorCoordinatorHolder#resetToCheckpoint runs during ExecutionGraph construction, before lazyInitialize supplies the scheduler executor, so the failJob underneath DeferrableCoordinator#resetAndStart throws from checkInitialized() inside an unobserved whenComplete callback and the job comes up RUNNING with a coordinator that never started. By start() the context is initialized, so the failure reaches context.failJob and terminates the job. SuppressRestartsException keeps the restart strategy from looping on a mismatch that would recur on every restore. Streaming reads are deliberately left unguarded: their checkpointed offset supersedes read.start-commit by design (IncrementalInputSplits), so validating it would break legitimate resumes. Bump the enumerator state serializer 1 -> 2 for the new fields, and record the nested split serializer's own version in the payload so the two formats can evolve independently instead of the outer version being forwarded to the split deserializer.
afd4780 to
1758313
Compare
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR makes a bounded (batch) HoodieSource fail fast when it resumes a checkpoint whose read.start-commit/read.end-commit range differs from the current config, by recording the range in the enumerator state (serializer bumped to VERSION 2, with V1 back-compat) and re-checking it on restore. I traced the guard logic, the V1/V2 serializer compatibility, the snapshotState override, and the Option-based bound comparison, and the commit-range mechanism itself looks correct. No new issues surfaced in this automated pass beyond the scope-completeness and deferral questions already raised inline in prior rounds. Please take a look at those existing inline comments, and this should be ready for a Hudi committer or PMC member to take it from here. A couple of small naming and clarity nits below.
| // The read.start-commit / read.end-commit bounds this bounded read was enumerated with, persisted | ||
| // in the enumerator checkpoint so a later restore can detect that they changed. See | ||
| // HoodieSource#checkBoundedCommitRangeUnchanged. | ||
| private final Option<String> readStartCommit; |
There was a problem hiding this comment.
🤖 nit: could you rename rangeFailure to rangeFailureMessage? The field holds a pre-formatted error string, and without that hint a reader has to look at the type (Option<String>) and the surrounding comment to figure out it's a message rather than, say, an exception or enum value.
| out.writeInt(splitStates.size()); | ||
| for (HoodieSourceSplitState splitState : splitStates) { | ||
| byte[] splitBytes = splitSerializer.serialize(splitState.getSplit()); | ||
| out.writeInt(splitBytes.length); |
There was a problem hiding this comment.
🤖 nit: the bare out.writeBoolean(false) here is hard to follow without a comment — could you add something like // lastEnumeratedInstantOffset: absent so a reader doesn't have to cross-reference the main serializer to know what field this writes?
| * {@code HoodieSource} compares these against the configured bounds and fails fast when they | ||
| * differ. See {@code HoodieSource#checkBoundedCommitRangeUnchanged}. | ||
| */ | ||
| Option<String> readStartCommit; |
There was a problem hiding this comment.
since we already have lastEnumeratedInstant and lastEnumeratedInstantOffset so that we can infer the start commit from the state recovery and we can just compare it agains the options from write config and throw if necessary, why snapshot the static read options every time in the state?
There was a problem hiding this comment.
For bounded reads there's nothing in those two to infer from — they're Option.empty() on every bounded checkpoint, and were before this PR too (HoodieStaticSplitEnumerator inherited AbstractHoodieSplitEnumerator:129; HoodieContinuousSplitEnumerator:90 is the only writer). So recovery can't reconstruct the configured range from state alone.
Repurposing the slots to carry the bounded bounds would avoid the version bump, but breaks on the reverse flip. HoodieSource:141 picks the enumerator from current config, not from what wrote the checkpoint, so a bounded checkpoint restored with read.streaming.enabled=true feeds those values into the resume position at :64. Both are completion times, so IncrementalInputSplits:266 would accept read.end-commit as startCompletionTime with OPEN_CLOSED and silently skip everything up to the old bound — this PR's bug in the other direction, and the shared domain is exactly what makes it silent instead of loud. Separate typed fields keep "streaming resume position" and "bounded scope assertion" from aliasing.
On cost: it's two optional strings per checkpoint, written only by the bounded enumerator, alongside the pending-split collection.
I also tried deriving from the splits' InstantRange — no new state at all — but IncrementalQueryAnalyzer:443-455 returns either no range or a null start bound when reading from earliest, and only incremental non-CDC splits carry one.
|
@danny0405 The offset approach doesn't carry over, unfortunately: DefaultHoodieSplitProvider.state() persists only pendingSplits, all stamped UNASSIGNED — assigned and completed splits aren't in the checkpoint at all. Streaming has one monotonic watermark that fully describes progress; a bounded read has a shrinking remainder set with no completion record, so there's no offset to resume from without adding a completed-split ledger. |
splits that already got assigned will also be checkpointed if it is not finished, and each split will remember the position it consumes to, which can be used for job recovery where the next run consumes from. As a user, my understanding is that I can declare the start/end commit for the bounded streaming(batch) job, and the job will recover automatically from the state to ensure consistency. I'm assuming the case we want to fix in this patch is not for job auto recovery, but a manual restart of the job with an optional flink state, and you want to validate the state and the user config consume range is consistent, is that right? |
@danny0405 Hi Danny, yes your understanding is 100% correct. I might misunderstand your earlier. But yes, the issue I observed is that when user restarts the same job in a different bound range without cleaning up checkpoints (previously reading 01/01/2026 but the job fails somehow and has checkpoints written, then restarts the job with config of 01/02/2026) the job would silently and wrongly resume reading from 01/01/2026. So I'm thinking of failing this new run with clear errors, since logging a warning could likely be unnoticeable. |
@ericyuan915 Thanks for clarifying. This sounds more like a job-upgrade compatibility check on the platform side than source recovery semantics. For example, when a new job version is deployed with an existing checkpoint/savepoint, the platform should compare the old and new job specifications. If options that affect the source consumption scope have changed, it should reject the upgrade or require the job to start without the old state. This is also how other FLIP-27 sources are generally structured. KafkaSource checkpoints the assigned topic partitions and their current/stopping offsets because they are the actual execution state required to resume the splits. On restore, those offsets take precedence over the newly configured There is an important distinction here: persisting a bound as part of a split is reasonable when the reader needs it to finish that restored split, as Kafka does with its stopping offset. Persisting static source options only as a configuration fingerprint for deployment-time validation is different. It couples job-upgrade validation to the source checkpoint format and requires the state schema to keep expanding whenever another option affects split enumeration. Therefore, I don't think recording |
5 participants
closes #19375
Describe the issue this Pull Request addresses
A bounded (batch) FLIP-27
HoodieSourceread restored from a checkpoint/savepoint reuses the persisted split set and never re-enumerates. So if you changeread.start-commit/read.end-commit, the query type, or the partition selection and then resume from that checkpoint, the job silently reads the checkpoint's old scope — no error, no warning. The split set is frozen at enumeration time and is not re-derived on restore, so the configuration change is silently ignored.Summary and Changelog
Fail fast on a changed bounded-read scope instead of silently reading a stale one:
read.start-commit/read.end-commitbounds, and the pruned partition set — and checkpoint it inHoodieSplitEnumeratorState(snapshotState). Projection /requiredColumnsis deliberately excluded: it changes what is read from each file, not which files (splits) are read.start()(wrapped inSuppressRestartsException) rather than thrown fromcreateEnumerator.createEnumeratorruns inside the coordinator's asynchronous reset, where a throw is swallowed and leaves a "zombie" job (RUNNING, no enumerator, no throughput).start()runs insiderunInEventLoop, whose uncaught-exception handler routes the throw tocontext.failJob;SuppressRestartsExceptionstops the restart strategy from looping on a mismatch that would otherwise recur on every restore.VERSION1 → 2. Checkpoints written by VERSION 1 carry no token and therefore restore unguarded (backward compatible).PartitionPruner#scopeToken()(default = class simple name) with concrete overrides inStaticPartitionPrunerand the chained pruner; both sort their contents so the token is order-independent.Tests added:
TestHoodieSourceScopeGuard— token reflects pruned partitions, changes when the scope widens (1 → 3 partitions), and is order-independent; unchanged scope → no failure; changed scope → actionable message; legacy (tokenless) checkpoint → no failure.TestHoodieEnumeratorStateSerializer— token round-trips; the legacy 3-arg constructor defaults the token to empty; VERSION 1 deserialization does not read a token.TestHoodieStaticSplitEnumerator—snapshotStatecarries the token;start()fails the job when the bounded scope changed.Impact
Bounded Flink reads that resume from a checkpoint/savepoint with a changed scope now fail fast instead of silently reading the old scope. This is a behavior change, but it only surfaces a case that was previously a silent data-correctness bug. Streaming reads are unaffected. Pre-existing checkpoints (serializer VERSION 1) restore exactly as before.
Risk Level
low.
start()behavior.Documentation Update
none. No new configs and no user-facing config/default changes; the new behavior only replaces a silent failure with an explicit, self-describing error message.
Contributor's checklist