singhpk234 commented on code in PR #14506:
URL: https://github.com/apache/iceberg/pull/14506#discussion_r2492604888
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java:
##########
@@ -296,6 +324,13 @@ private Snapshot latestSnapshot(Table table, String
branch) {
return table.snapshot(branch);
}
+ private Snapshot latestSnapshot(TableMetadata metadata, String branch) {
+ if (branch == null) {
+ return metadata.currentSnapshot();
+ }
+ return metadata.snapshot(metadata.ref(branch).snapshotId());
+ }
Review Comment:
probably we can move this to `SnapshotUtil`
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java:
##########
@@ -315,6 +350,25 @@ private Map<Integer, Long>
lastCommittedOffsetsForTable(Table table, String bran
return ImmutableMap.of();
}
+ private Map<Integer, Long> lastCommittedOffsetsForTable(TableMetadata
metadata, String branch) {
+ Snapshot snapshot = latestSnapshot(metadata, branch);
+ while (snapshot != null) {
Review Comment:
[optional] we can get an iterable of ancestors from the SnapshotUtils
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java:
##########
@@ -284,6 +289,29 @@ private void commitToTable(
}
}
+ private void validateAndCommit(
+ PendingUpdate<?> pendingUpdate, String branch, Map<Integer, Long>
expectedOffsets) {
+ CommitValidator validator =
+ (base, metadata) -> {
+ Map<Integer, Long> lastCommittedOffsets =
lastCommittedOffsetsForTable(base, branch);
+
+ if (expectedOffsets == null || expectedOffsets.isEmpty()) {
+ return; // there are no stored offsets, so assume we're starting
with new offsets
+ }
+
+ if (!expectedOffsets.equals(lastCommittedOffsets)) {
+ throw new CommitFailedException(
+ "Latest offsets do not match expected offsets for this
commit.");
+ }
+ };
Review Comment:
[doubt] wouldn't this be lastCommittedOffset ?
if expectedOffset is null and lastCommittedOffset is non empty then we
should fail ?
```suggestion
Map<Integer, Long> lastCommittedOffsets =
lastCommittedOffsetsForTable(base, branch);
if (lastCommittedOffsets == null ||
lastCommittedOffsets.isEmpty()) {
return; // there are no stored offsets, so assume we're starting
with new offsets
}
// handle case for expectedOffsets being null too
if (!lastCommittedOffsets.equals(expectedOffsets)) {
throw new CommitFailedException(
"Committed offsets do not match expected offsets for this
commit.");
}
};
```
--
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]
