Validate parent snapshots in Kafka Connect #14515

aiborodin · 2025-11-06T07:20:52Z

This change uses the new API in #14514 to validate that no concurrent commit has moved the Kafka offsets that would not be seen by the committer due to the table refreshing during the commit process.

It is an alternative solution to #14510.

Change-Id: Ie7e2bc0920bd855188c5bcc2435ace947db9af21

Change-Id: Ib7c367e90f42a8184772fb17194bb4e5427804db

aiborodin · 2025-11-06T22:56:56Z

@rdblue What do you think?

core/src/test/java/org/apache/iceberg/TestSnapshotProducer.java

rdblue · 2025-11-06T23:44:48Z

core/src/main/java/org/apache/iceberg/SnapshotProducer.java

  private TableMetadata base;
  private boolean stageOnly = false;
  private Consumer<String> deleteFunc = defaultDelete;
+  @Nullable private Long startingSnapshotId = null; // check all versions by default


I like the original PR's approach to iterate backward through history rather than this approach that iterates through more by default. I also like that the other approach can reuse work by running the validation once with a loop, rather than calling the validation in a loop.

I like the original PR's approach to iterate backward through history rather than this approach that iterates through more by default.

Both the original PR and this approach iterate backward. The original PR loops through the whole history for every validation run. This PR only checks new parent Snapshots.

I also like that the other approach can reuse work by running the validation once with a loop, rather than calling the validation in a loop.

Like I said above - the other approach runs validations for all parent snapshots in a loop. This PR limits the scope of validations to new Snapshots.

rdblue · 2025-11-06T23:46:05Z

core/src/main/java/org/apache/iceberg/SnapshotProducer.java

  private boolean stageOnly = false;
  private Consumer<String> deleteFunc = defaultDelete;
+  @Nullable private Long startingSnapshotId = null; // check all versions by default
+  @Nullable private Consumer<Snapshot> snapshotValidator = null;


As I commented on the original PR, we don't want to leave the exception class to the implementation. A validation should return a boolean to indicate that something was wrong. That way, the exception class is uniform (ValidationException) and acceptance is explicit.

rdblue · 2025-11-06T23:47:22Z

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java

+      operation.set(snapshotOffsetsProp, offsetsJson);
+      operation.set(COMMIT_ID_SNAPSHOT_PROP, commitState.currentCommitId().toString());
+      if (validThroughTs != null) {
+        operation.set(VALID_THROUGH_TS_SNAPSHOT_PROP, validThroughTs.toString());


It is generally better to avoid unnecessary changes and refactoring. If you want to refactor this to remove the duplication, then we prefer opening a separate PR.

Agreed. Unfortunately, it resulted in the cyclomatic complexity failure, so I had to change it.

rdblue · 2025-11-06T23:49:23Z

core/src/test/java/org/apache/iceberg/TestSnapshotProducer.java

+    @Override
+    public void accept(Snapshot snapshot) {
+      if (stagedWapId.equals(snapshot.summary().get(SnapshotSummary.STAGED_WAP_ID_PROP))) {
+        throw new ValidationException(validationErrorMessage);


Minor: Use ValidationException.check instead.

In addition, exception messages should have all relevant context. In this case, I'd expect it to contain the WAP ID. I know this is just test code, but it seems like a good rule to follow in all cases.

rdblue · 2025-11-06T23:50:43Z

core/src/test/java/org/apache/iceberg/TestSnapshotProducer.java

+
+    private WapIdValidator(String stagedWapId, String validationErrorMessage) {
+      this.stagedWapId = stagedWapId;
+      this.validationErrorMessage = validationErrorMessage;


This is a test, but I would not want people looking at this to copy the convention used here of passing an error message into the validator. I'd expect this to be embedded in the exception produced by validation. This is probably related to the issue below, where the actual ID is missing.

rdblue · 2025-11-06T23:52:43Z

core/src/main/java/org/apache/iceberg/StreamingDelete.java


  @Override
  protected void validate(TableMetadata base, Snapshot parent) {
+    super.validate(base, parent);


I considered whether it is a good idea to override validate and concluded that I wouldn't recommend this option. Part of the problem is that this needlessly touches a lot of files it doesn't need to, but it is also more brittle than adding a call because we may miss an implementation. And there are some custom operations out there that extend Iceberg classes that this would break.

rdblue · 2025-11-06T23:54:14Z

api/src/main/java/org/apache/iceberg/SnapshotUpdate.java

+   * @param snapshotValidator a user function to validate parent snapshots
+   * @return this for method chaining
+   */
+  default ThisT validateWith(Consumer<Snapshot> snapshotValidator) {


I think the use of the ancestor iterable is a more elegant solution than making a large refactor and calling the validator in a loop.

rdblue · 2025-11-06T23:55:21Z

core/src/main/java/org/apache/iceberg/BaseReplacePartitions.java

      if (dataSpec().isUnpartitioned()) {
        validateAddedDataFiles(
-            currentMetadata, startingSnapshotId, Expressions.alwaysTrue(), parent);
+            currentMetadata, startingSnapshotId(), Expressions.alwaysTrue(), parent);


This is a good example of what I mean in the comment above. There are quite a few changes here (and in other classes) to refactor so that startingSnapshotId can be reused.

rdblue · 2025-11-06T23:58:21Z

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java

+
+      ValidationException.check(
+          expectedOffsets.equals(lastCommittedOffsets),
+          "Latest offsets do not match expected offsets for this commit. Table: %s, Expected: %s, Last Committed: %s",


Error messages in Iceberg aim to be direct and simple, rather than using full sentences and punctuation. In this case, I'd expect something more like "Cannot commit to %s: expected offsets %s changed to %s"

rdblue · 2025-11-07T00:03:37Z

api/src/main/java/org/apache/iceberg/SnapshotUpdate.java

+
+  /**
+   * Enables snapshot validation with a user-provided function, which must throw a {@link
+   * org.apache.iceberg.exceptions.ValidationException} on validation failures.


I think this should document how the validator is called. Is the latest snapshot first or last? I think that affects how validations are written so it seems important.

aiborodin · 2025-11-07T04:28:58Z

Thank you for the review @rdblue, I appreciate your feedback!
Given that #14509 has been merged, we should probably go ahead with that API.
I still think there's value in adding validateFromSnapshot(long snapshotId) to allow clients to limit the checked history. I can raise a separate PR for this feature.

github-actions · 2025-12-08T00:19:09Z

This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the [email protected] list. Thank you for your contributions.

github-actions · 2025-12-15T00:20:36Z

This pull request has been closed due to lack of activity. This is not a judgement on the merit of the PR in any way. It is just a way of keeping the PR queue manageable. If you think that is incorrect, or the pull request requires review, you can revive the PR at any time.

aiborodin added 2 commits November 6, 2025 17:26

Validate parent snapshots

d9c70d1

Change-Id: Ie7e2bc0920bd855188c5bcc2435ace947db9af21

Validate committed offsets in Kafka connect

d992701

Change-Id: Ib7c367e90f42a8184772fb17194bb4e5427804db

github-actions bot added API core KAFKACONNECT labels Nov 6, 2025

This was referenced Nov 6, 2025

Add SnapshotUpdateValidator to validate snapshots on commit #14509

Merged

Implement Snapshot validation API for commits #14514

Closed

Kafka Connect: validate offsets for refreshed table state on commit #14510

Merged

rdblue reviewed Nov 6, 2025

View reviewed changes

core/src/test/java/org/apache/iceberg/TestSnapshotProducer.java Show resolved Hide resolved

rdblue reviewed Nov 6, 2025

View reviewed changes

rdblue reviewed Nov 7, 2025

View reviewed changes

github-actions bot added the stale label Dec 8, 2025

github-actions bot closed this Dec 15, 2025

Validate parent snapshots in Kafka Connect #14515

Validate parent snapshots in Kafka Connect #14515

Uh oh!

Conversation

aiborodin commented Nov 6, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

aiborodin commented Nov 6, 2025

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

aiborodin commented Nov 7, 2025

Uh oh!

github-actions bot commented Dec 8, 2025

Uh oh!

github-actions bot commented Dec 15, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

aiborodin commented Nov 6, 2025 •

edited

Loading