mxm commented on code in PR #18807:
URL: https://github.com/apache/iceberg/pull/18807#discussion_r4515197369
##########
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:
NIT: `DynamicCommittable` adds "Committables restored from older state use
the jobId as sinkId." Could we add it here as well? `deserializeV1` passes the
jobId as sinkId without a comment. Maybe also add a comment there.
##########
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:
Could we add the reverse case: this legacy snapshot with a committable that
has the same jobId but a new sinkId (stateless restart right after upgrading)?
--
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]
