danny0405 commented on issue #19763: URL: https://github.com/apache/hudi/issues/19763#issuecomment-5436621858
Thanks for the detailed reproduction. I agree that the physical overwrite follows from the first-record-dependent handle selection, but I do not think unaligned-checkpoint recovery is supported by the current Flink Hudi sink protocol. The 0.15.0 writer documents its exactly-once model as buffering data between checkpoints and keeping one Hudi instant within one checkpoint: https://github.com/apache/hudi/blob/release-0.15.0/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteFunction.java#L63-L90. When snapshotState runs, it flushes the current buckets: https://github.com/apache/hudi/blob/release-0.15.0/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteFunction.java#L132-L138, then sends the last-batch event and clears the buckets and write handles: https://github.com/apache/hudi/blob/release-0.15.0/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteFunction.java#L450-L490. That protocol implicitly relies on an aligned barrier after the file-ID shuffle, so all pre-barrier input has reached the writer before snapshotState closes the checkpoint batch. An unaligned checkpoint deliberately lets the barrier overtake in-flight channel data and restores that data before accepting new upstream input: https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/concepts/stateful-stream-processing/#unaligned-checkpointing. Consequently, a restored record can have an assignment produced before the checkpoint but arrive at the Hudi writer after the corresponding buckets and handles have been cleared and a new instant has started. The INSERT-first behavior is therefore a manifestation of the larger checkpoint-boundary mismatch. The handle factory does indeed select create versus merge from the first record: https://github.com/apache/hudi/blob/release-0.15.0/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkWriteHandleFactory.java#L97-L126. Changing only that selection may prevent this particular base-file hiding pattern, but it would not by itself establish correct exactly-once semantics for replayed channel state. For now, execution.checkpointing.unaligned.enabled=false is the correct workaround. I would keep this issue as a valid safety/support gap: Hudi should at minimum document or reject this configuration so an unsupported recovery mode cannot silently corrupt a table. Supporting unaligned recovery would require making the instant/bucket protocol account for restored channel state, not only changing the write handle chosen for an existing file group. -- 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]
