UrsSchoenenbergerNu commented on issue #15846: URL: https://github.com/apache/iceberg/issues/15846#issuecomment-4243772004
Hi @mxm , at first glance, I don't think IcebergSink V2 is affected by this particular bug: FlinkSink in its current state has an internal tracking of records per checkpoint (in IcebergFilesCommitter) that is not correctly "aligned" because with unaligned checkpoints, records may arrive for N after the snapshot has already been taken. IcebergSink seems to, in IcebergWriteAggregator, use the checkpoint ID at snapshot time (in preprareSnapshotPreBarrier), so records arriving afterwards would always be part of N+1 from its point of view. CommitableMessages for N should therefore only be emitted once and should be consistent per-checkpoint even in recovery cases. IcebergCommitter::commit also uses a per-checkpoint map - but it's not persistent state, and it uses the above notion of checkpoint ID which should be safe? But I'm not an expert on how SinkV2 and Committables work internally, so I'm only 50% confident that this explanation makes sense. I don't think I can run my test harness against IcebergSink, as the test harness that allows controlling the timing of snapshostState() only works against a single operator, whereas if the bug manifested on V2, it would likely be on the combination of IcebergWriteAggregator->IcebergCommitter? Please let me know what you think. (As an aside, the reason we're not using v2 IcebergSink in the first place is the warning in the docs that it's still considered experimental.) -- 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] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
