danny0405 opened a new pull request, #20516: URL: https://github.com/apache/hudi/pull/20516
### Describe the issue this Pull Request addresses Issue: #20407, following the [Phase 3 request](https://github.com/apache/hudi/issues/20401#issuecomment-6599774764). This is separate from the basic chaining acceptance in #20515. A downstream Hudi streaming table can retain stale rows or read misleading increments after its upstream table is overwritten/recreated or required source history expires. Add an opt-in source guard that requires an explicit downstream table/checkpoint rebuild before continuing. ### Summary and Changelog - Add `hoodie.datasource.read.streaming.fail.on.data.loss` (default `false`) to both streaming source offset versions. Validate the saved offset against the active completed timeline, cleaner retention metadata, and non-clustering replace commits before constructing the batch. - Reject missing/archived checkpoint history, physical recreation, backwards offsets, and logical overwrites, including resets completed after the captured batch end. Preserve clustering, compaction, and fresh earliest reads for recovery. - Capture the flag at source construction through the existing `spark.hoodie.*` SQLConf helper. Direct reader options take precedence. Spark 4.1.1 discards reader options on graph-internal references, so SDP must enable the flag through pipeline/flow SQLConf. - Add six source regression cases covering COW/MOR, requested/completion offsets, clustering, overwrite, physical recreation, missing active history, cleaner boundaries, and configuration precedence. - Add a CLI/Connect reset-and-rebuild acceptance scenario and recovery documentation. The Python scenario is adapted from the chaining acceptance in #20515 and uses the external #20453 harness; production guard logic is new. ### Impact Opt-in latest-state streaming behavior; the default and CDC reads retain existing behavior. No storage-format or checkpoint-format changes. Enabled readers reload active history and read cleaner/replace metadata per batch. This deliberately requires an active checkpoint instant: archival alone can require a rebuild even when archived metadata still exists. Destructive changes must be coordinated with consumers because validation and execution are not atomic. Recreation is detected on the next resumed batch, and copying identical timeline history is not a new identity check. Hard-delete CDC and bounded AvailableNow admission control remain outside this change. ### Risk Level medium Changes affect batch admission in both streaming sources, with conservative rejection possible when the flag is enabled. Validation on JDK 17 / Spark 4.1 / Scala 2.13: - Reactor `mvn install -Pjava17 -Dspark4.1 -Dscala-2.13 -DskipTests -Dgpg.skip -Djacoco.skip -am -pl packaging/hudi-spark-bundle`: passed, including compilation/style checks. - `TestStreamingSource`: all 22 tests passed. - Spark 4.1.1 CLI/Connect acceptance: COW and MOR passed with metadata disabled. Verified exact incremental rows, unchanged restart timelines, recreation rejection before any downstream commit, and exact rebuilt snapshots after truncating both tables and assigning fresh pipeline checkpoints. MOR exercised compaction. - Python compilation and `git diff --check`: passed. The CLI `--full-refresh-all` path still fails on the base branch's catalog `DeleteColumn` handling (tracked by #20492). The acceptance scenario verifies explicit coordinated recovery; it does not claim CLI full-refresh or metadata-enabled acceptance coverage. Other Spark profiles were not run. ### Documentation Update Added the config description and `hudi-integ-test/spark-pipelines/SOURCE_SAFETY.md`, linked from `ACCEPTANCE.md`, covering configuration propagation, retained-history requirements, recovery, and limitations. The corresponding Hudi website documentation on `asf-site` remains a follow-up before release; this PR does not claim that the website was updated. ### Contributor's checklist - [x] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute) - [x] Enough context is provided in the sections above - [x] Adequate tests were added if applicable -- 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]
