danielcweeks commented on code in PR #14510:
URL: https://github.com/apache/iceberg/pull/14510#discussion_r2496385914
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java:
##########
@@ -284,6 +290,25 @@ private void commitToTable(
}
}
+ private SnapshotUpdateValidator offsetValidator(
+ TableIdentifier tableIdentifier, Map<Integer, Long> expectedOffsets) {
+
+ return (baseSnapshots, updatedSnapshots) -> {
Review Comment:
I feel like there's an issue if you can't distinguish the `base` updates
from the ones that are already applied, which is both are included. The KC
implementation doesn't care about the updated state because we're largely
validating that the state we're building off didn't change.
You could update this to just look at the updated and then skip first set of
offsets (those will be the one we are attempting to commit) and then look back
the the next one. But it seems more awkward because you have to be aware of the
context of the changes you already applied (as opposed to being able to check
the original state).
--
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]
