nsivabalan opened a new issue, #20530:
URL: https://github.com/apache/hudi/issues/20530
### Task Description
Support more than one SDP streaming flow writing to the same Hudi streaming
table (fan-in) with exactly-once batch dedup per flow. #20406 rejects fan-in
for the initial scope. This task removes that restriction.
**How the sink dedups today.** `HoodieStreamingSink` writes
`_hudi_streaming_sink_checkpoint = {<identifier>: <batchId>}` into each
commit's metadata (`CommitUtils.getCheckpointValueAsString`). It skips an
incoming batch if the latest commit carrying its identifier has a batch id ≥
the incoming one (`canSkipBatch`, via
`CommitUtils.getValidCheckpointForCurrentWriter`).
**Gaps for fan-in:**
1. **Shared default identifier.** The identifier defaults to
`default_single_writer`, and SDP doesn't set a per-flow one. Two flows into one
table compare their batch ids against each other's watermark, so flow B's batch
3 is skipped once flow A has committed batch 40. That's silent data loss.
2. **Each commit stores only its own writer's entry.** The map holds one
key, so a lookup walks back the active timeline until it finds a commit from
the same writer.
- The cost grows with the other flows' commit rate.
- If one flow is idle long enough for its last commit to be archived, the
lookup finds nothing. After a restart the sink falls back to its in-memory
watermark (-1) and re-applies a batch that was already committed. That's
duplicates on insert/append targets.
3. **Concurrency.** Fan-in means concurrent writers to one table within a
pipeline run. The sink path needs a supported concurrency setup (OCC/NBCC with
a lock provider). It also needs a defined behaviour when a batch loses a write
conflict: retry the same batch id, never skip it.
**Proposed direction (open for discussion):**
- Derive a stable per-flow identifier automatically (for example from flow
name plus checkpoint generation). This is aligned with the
checkpoint-generation scoping in #20404. An explicit
`hoodie.datasource.write.streaming.checkpoint.identifier` keeps working as
today.
- Carry forward every writer's entry: at commit time, merge the latest
committed checkpoint map with this flow's entry, so each commit holds `{flowA:
n, flowB: m, ...}`. A lookup then only needs the latest commit carrying the
key. This is similar in spirit to how HoodieStreamer carries checkpoint keys
forward via `hoodie.write.meta.key.prefixes`. The merge must read the latest
map inside the commit's conflict-resolution window so concurrent flows don't
drop each other's entries.
- Decide how archived or legacy single-entry metadata is read during upgrade.
### Acceptance criteria
- [ ] Two flows writing one Hudi ST each commit all of their batches.
Neither flow's batches are skipped because of the other flow's batch ids.
- [ ] Restarting either flow, including after interruption between the Hudi
commit and the Spark commit log, does not re-apply that flow's committed
batches.
- [ ] An idle flow whose last commit has been archived still dedups
correctly after a restart.
- [ ] Concurrent commits from the two flows are conflict-checked. A
conflicting batch is retried, not skipped.
- [ ] Full refresh of the target resets every flow's watermark consistently
with #20404.
- [ ] The fan-in rejection from #20406 is lifted, and the docs describe the
required concurrency configuration.
- [ ] CLI/Connect acceptance covers two flows into one target (COW and MOR,
metadata table enabled).
### Task Type
Code improvement/refactoring
### Related Issues
Parent feature issue: #20401
Related: #20406 (initial fan-in rejection), #20404 (checkpoint-generation
isolation), #20378
--
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]
