mxm commented on code in PR #13714:
URL: https://github.com/apache/iceberg/pull/13714#discussion_r2348134707
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergWriteAggregator.java:
##########
@@ -76,11 +85,21 @@ public void open() throws Exception {
@Override
public void finish() throws IOException {
- prepareSnapshotPreBarrier(Long.MAX_VALUE);
+ prepareSnapshotPreBarrier(lastCheckpointId + 1);
}
@Override
public void prepareSnapshotPreBarrier(long checkpointId) throws IOException {
+ if (checkpointId == lastCheckpointId) {
+ // Already flushed. This can happen when finish() above triggers
flushing prior creating the
+ // final checkpoint. The calls are mutually exclusive, but we need to
ensure we don't flush
Review Comment:
>can the Flink trigger multiple checkpoints after the finish() call?
Yes. Technically, I suppose we could skip the empty comittables, but we have
been doing them so far, so not sure it is an issue.
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergWriteAggregator.java:
##########
@@ -76,11 +85,21 @@ public void open() throws Exception {
@Override
public void finish() throws IOException {
- prepareSnapshotPreBarrier(Long.MAX_VALUE);
+ prepareSnapshotPreBarrier(lastCheckpointId + 1);
}
@Override
public void prepareSnapshotPreBarrier(long checkpointId) throws IOException {
+ if (checkpointId == lastCheckpointId) {
+ // Already flushed. This can happen when finish() above triggers
flushing prior creating the
+ // final checkpoint. The calls are mutually exclusive, but we need to
ensure we don't flush
+ // twice.
+ LOG.debug("Aggregated writes for checkpoint id {} already flushed.",
checkpointId);
Review Comment:
Ack, changing.
--
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]
