Guosmilesmile commented on code in PR #18807:
URL: https://github.com/apache/iceberg/pull/18807#discussion_r4524468513
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergCommitter.java:
##########
@@ -315,6 +319,37 @@ public void
testCommitTxnSkipsAlreadyCommittedCheckpointAfterRestore() throws Ex
SimpleDataUtil.assertTableRows(table,
ImmutableList.of(rowFromPreviousRun), branch);
}
+ @TestTemplate
+ public void testSkipsAlreadyCommittedOlderCommittable() throws Exception {
+ RowData rowFromPreviousRun = SimpleDataUtil.createRowData(1, "hello1");
+ DataFile previousRunFile =
+ writeDataFile("data-previous-run",
ImmutableList.of(rowFromPreviousRun));
+
+ // Snapshot written before the sinkId was introduced
+ table
+ .newAppend()
+ .appendFile(previousRunFile)
+ .set(SinkUtil.FLINK_JOB_ID, jobId)
Review Comment:
Add a ut testCommitsWithNewSinkIdAfterUpgrade for this case.
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergCommittable.java:
##########
@@ -26,20 +26,23 @@
/**
* The aggregated results of a single checkpoint which should be committed.
Containing the
* serialized {@link org.apache.iceberg.flink.sink.DeltaManifests} file -
which contains the commit
- * data, and the jobId, operatorId, checkpointId triplet which helps
identifying the specific commit
+ * data, and the jobId, sinkId, operatorId, checkpointId which identify the
specific commit
Review Comment:
Ok, add comment for it.
--
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]
