PDGGK commented on PR #19776:
URL: https://github.com/apache/hudi/pull/19776#issuecomment-5584347274
Fair concern — I went and checked rather than argue from the diff. Three
things, all from the current code:
**1. The coalesce path throws the key away in the same stage.**
`SparkStreamingMetadataWriteHandler`:
```java
.mapToPair(ws -> new Tuple2(ws.getStat().getPath(), ws))
.partitionBy(new CoalescingPartitioner(coalesceParallelism))
.map(entry -> entry._2) // <- key gone here
```
The result goes straight into `.union(mdtWriteStatuses)`. Nothing downstream
reads which task a `WriteStatus` landed on, and nothing in
`hudi-client/hudi-spark-client/.../client/` reads
`TaskContext.getPartitionId()` at all.
**2. There is no cross-run routing stability today, so there is none to
break.** `coalesceParallelism` is `Math.max(1,
dataTableWriteStatuses.getNumPartitions() / coalesceDivisorForDataTableWrites)`
— derived from *this run's* upstream partition count. `outputSparkPartitions`
in the three bulk-insert callers is the same shape. Change the input size and
every key reroutes, with or without this patch. A restart that re-derives a
different parallelism already moves records between tasks.
**3. The state is per-instant, not per-stream.**
`StreamingMetadataWriteHandler` keeps `{instant -> metadata writer}` and says
it is "cleaned up when action is completed or when write client is closed".
Despite the name it is not a Structured Streaming checkpoint — Flink has its
own subclass of the same handler. And the three bulk-insert callers all end in
`.values()`, so the only invariant they need is that one partition path lands
in one Spark partition, which any deterministic function of the hash gives.
So I do not think a restart can see a stale routing decision: there is
nowhere for one to be written down.
What I have *not* done is drive an actual restart end to end — I checked the
code paths, not a running job. If you would rather see that than take the
reading, say so and I will add a test that writes with one parallelism,
restarts, and writes with another, asserting the data table and MDT agree.
Happy to leave this parked until you are comfortable either way.
--
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]
