mxm commented on code in PR #18303:
URL: https://github.com/apache/iceberg/pull/18303#discussion_r4272212746
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java:
##########
@@ -166,12 +171,22 @@ public void
commit(Collection<CommitRequest<DynamicCommittable>> commitRequests)
jobEntry : jobEntries) {
JobOperatorKey jobKey = jobEntry.getKey();
long maxCommittedCheckpointId =
- getMaxCommittedCheckpointId(ancestors, jobKey.jobId(),
jobKey.operatorId());
+ isRestored
+ ? getMaxCommittedCheckpointId(ancestors, jobKey.jobId(),
jobKey.operatorId(), null)
+ : INITIAL_CHECKPOINT_ID;
Review Comment:
Imagine this scenario:
1. A table where the maxCommittedCheckpointId=1000
2. Stateless restart resets the checkpoint id to 1
3. We fail to commit checkpoint 1 (e.g. catalog permission issue)
=> We will drop data for all other 999 checkpoints because the restart is
stateful and isRestored==true.
I raised this already in https://github.com/apache/iceberg/pull/18123.
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java:
##########
@@ -126,6 +126,7 @@ public Committer<DynamicCommittable>
createCommitter(CommitterInitContext contex
flinkWriteConf.overwriteMode(),
flinkWriteConf.workerPoolSize(),
sinkId,
+ context.getRestoredCheckpointId().isPresent(),
Review Comment:
In batch mode, getRestoredCheckpointId() is always empty (=>
isRestored==false). This means that if we fail a batch commit the retry can
result in duplicate data.
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java:
##########
@@ -181,17 +196,29 @@ public void
commit(Collection<CommitRequest<DynamicCommittable>> commitRequests)
jobEntry.getValue().tailMap(maxCommittedCheckpointId, false);
if (!uncommitted.isEmpty()) {
commitPendingRequests(
- table, tableKey.branch(), uncommitted, jobKey.jobId(),
jobKey.operatorId());
+ table,
+ tableKey.branch(),
+ uncommitted,
+ jobKey.jobId(),
+ jobKey.operatorId(),
+ latestSnapshotId);
}
}
}
}
private static long getMaxCommittedCheckpointId(
- Iterable<Snapshot> ancestors, String flinkJobId, String operatorId) {
+ Iterable<Snapshot> ancestors,
+ String flinkJobId,
+ String operatorId,
+ @Nullable Long stopAtSnapshotId) {
long lastCommittedCheckpointId = INITIAL_CHECKPOINT_ID;
for (Snapshot ancestor : ancestors) {
+ if (stopAtSnapshotId != null && stopAtSnapshotId ==
ancestor.snapshotId()) {
+ break;
+ }
Review Comment:
Thanks! The code here is confusing. All this `stopAtSnapshotId` does is to
stop after the latest snapshot and return -1 as the checkpoint id. This
indirection is not exactly obvious. There is also no table refresh that can
happen during passing down the snapshot id.
--
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]
