REST: Reconcile on CommitStateUnknown for simple update (#14320)

* Reconcile on CommitStateUnknown for simple update

* fix check style error

* fix checking style error

* address comments

* address comments

* add reconcileOnSimpleUpdate method

* remove config

* remove empty line

* remove un-needed test

* add comment

* spotlessApply

* add TestCommitStateUnknownNotReconciled

Author: huaxingao

core/src/main/java/org/apache/iceberg/rest/RESTTableOperations.java

@@ -26,12 +26,14 @@
 import java.util.function.Supplier;
 import org.apache.iceberg.LocationProviders;
 import org.apache.iceberg.MetadataUpdate;
+import org.apache.iceberg.SnapshotRef;
 import org.apache.iceberg.TableMetadata;
 import org.apache.iceberg.TableOperations;
 import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.UpdateRequirement;
 import org.apache.iceberg.UpdateRequirements;
 import org.apache.iceberg.encryption.EncryptionManager;
+import org.apache.iceberg.exceptions.CommitStateUnknownException;
 import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.io.LocationProvider;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
@@ -155,20 +157,90 @@ public void commit(TableMetadata base, TableMetadata metadata) {
     // the error handler will throw necessary exceptions like CommitFailedException and
     // UnknownCommitStateException
     // TODO: ensure that the HTTP client lib passes HTTP client errors to the error handler
-    LoadTableResponse response =
-        client.post(path, request, LoadTableResponse.class, headers, errorHandler);
+    LoadTableResponse response;
+    try {
+      response = client.post(path, request, LoadTableResponse.class, headers, errorHandler);
+    } catch (CommitStateUnknownException e) {
+      // Lightweight reconciliation for snapshot-add-only updates on transient unknown commit state
+      if (updateType == UpdateType.SIMPLE && reconcileOnSimpleUpdate(updates, e)) {
+        return;
+      }
+
+      throw e;
+    }
 
     // all future commits should be simple commits
     this.updateType = UpdateType.SIMPLE;
 
     updateCurrentMetadata(response);
   }
 
+  /**
+   * Attempt best-effort reconciliation for SIMPLE updates that only add a snapshot.
+   *
+   * <p>Returns true if the expected snapshot is observed in the refreshed table state. Returns
+   * false if the expected snapshot cannot be determined, is not present after refresh, or if the
+   * refresh fails. In case of refresh failure, the failure is recorded as suppressed on the
+   * provided {@code original} exception to aid diagnostics.
+   */
+  private boolean reconcileOnSimpleUpdate(
+      List<MetadataUpdate> updates, CommitStateUnknownException original) {
+    Long expectedSnapshotId = expectedSnapshotIdIfSnapshotAddOnly(updates);
+    if (expectedSnapshotId == null) {
+      return false;
+    }
+
+    try {
+      TableMetadata refreshed = refresh();
+      return refreshed != null && refreshed.snapshot(expectedSnapshotId) != null;
+    } catch (RuntimeException reconEx) {
+      original.addSuppressed(reconEx);
+      return false;
+    }
+  }
+
   @Override
   public FileIO io() {
     return io;
   }
 
+  private static Long expectedSnapshotIdIfSnapshotAddOnly(List<MetadataUpdate> updates) {
+    Long addedSnapshotId = null;
+    Long mainRefSnapshotId = null;
+
+    for (MetadataUpdate update : updates) {
+      if (update instanceof MetadataUpdate.AddSnapshot) {
+        if (addedSnapshotId != null) {
+          return null; // multiple snapshot adds -> not safe
+        }
+        addedSnapshotId = ((MetadataUpdate.AddSnapshot) update).snapshot().snapshotId();
+      } else if (update instanceof MetadataUpdate.SetSnapshotRef) {
+        MetadataUpdate.SetSnapshotRef setRef = (MetadataUpdate.SetSnapshotRef) update;
+        if (!SnapshotRef.MAIN_BRANCH.equals(setRef.name())) {
+          return null; // only allow main ref update
+        }
+        mainRefSnapshotId = setRef.snapshotId();
+      } else {
+        // any other update type makes this not a pure snapshot-add
+        return null;
+      }
+    }
+
+    if (addedSnapshotId == null) {
+      return null;
+    }
+
+    if (mainRefSnapshotId != null && !addedSnapshotId.equals(mainRefSnapshotId)) {
+      // Only handle "append to main" here. In this request, main is being set to a snapshot ID
+      // that is different from the snapshot we just added (e.g., rollback or move main elsewhere).
+      // In that case, finding the added snapshot in history doesn't tell us whether main moved to
+      // it, so skip reconciliation.
+      return null;
+    }
+
+    return addedSnapshotId;
+  }
+
   private TableMetadata updateCurrentMetadata(LoadTableResponse response) {
     // LoadTableResponse is used to deserialize the response, but config is not allowed by the REST
     // spec so it can be

core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java

@@ -65,6 +65,7 @@
 import org.apache.iceberg.catalog.TableCommit;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.exceptions.CommitFailedException;
+import org.apache.iceberg.exceptions.CommitStateUnknownException;
 import org.apache.iceberg.exceptions.NotAuthorizedException;
 import org.apache.iceberg.exceptions.NotFoundException;
 import org.apache.iceberg.exceptions.ServiceFailureException;
@@ -2900,6 +2901,101 @@ public <T extends RESTResponse> T execute(
     return catalog(adapter);
   }
 
+  @Test
+  public void testReconcileOnUnknownSnapshotAddMatchesSnapshotId() {
+    RESTCatalogAdapter adapter = Mockito.spy(new RESTCatalogAdapter(backendCatalog));
+
+    RESTCatalog catalog =
+        new RESTCatalog(SessionCatalog.SessionContext.createEmpty(), (config) -> adapter);
+    catalog.initialize(
+        "test",
+        ImmutableMap.of(
+            CatalogProperties.FILE_IO_IMPL, "org.apache.iceberg.inmemory.InMemoryFileIO"));
+
+    if (requiresNamespaceCreate()) {
+      catalog.createNamespace(TABLE.namespace());
+    }
+
+    catalog.createTable(TABLE, SCHEMA);
+
+    // Simulate: server commits, but client receives CommitStateUnknown (transient 5xx)
+    Mockito.doAnswer(
+            invocation -> {
+              invocation.callRealMethod();
+              throw new CommitStateUnknownException(
+                  new ServiceFailureException("Service failed: 503"));
+            })
+        .when(adapter)
+        .execute(
+            reqMatcher(HTTPMethod.POST, RESOURCE_PATHS.table(TABLE)),
+            eq(LoadTableResponse.class),
+            any(),
+            any());
+
+    Table table = catalog.loadTable(TABLE);
+
+    // Perform a snapshot-adding commit; should reconcile instead of failing
+    table.newFastAppend().appendFile(FILE_A).commit();
+
+    // Extract the snapshot id we attempted to commit from the request body
+    long expectedSnapshotId =
+        allRequests(adapter).stream()
+            .filter(
+                r -> r.method() == HTTPMethod.POST && r.path().equals(RESOURCE_PATHS.table(TABLE)))
+            .map(HTTPRequest::body)
+            .filter(UpdateTableRequest.class::isInstance)
+            .map(UpdateTableRequest.class::cast)
+            .map(
+                req ->
+                    (MetadataUpdate.AddSnapshot)
+                        req.updates().stream()
+                            .filter(u -> u instanceof MetadataUpdate.AddSnapshot)
+                            .findFirst()
+                            .orElseThrow())
+            .map(add -> add.snapshot().snapshotId())
+            .findFirst()
+            .orElseThrow();
+
+    Table reloaded = catalog.loadTable(TABLE);
+    assertThat(reloaded.currentSnapshot()).isNotNull();
+    assertThat(reloaded.snapshot(expectedSnapshotId)).isNotNull();
+  }
+
+  @Test
+  public void testCommitStateUnknownNotReconciled() {
+    RESTCatalogAdapter adapter = Mockito.spy(new RESTCatalogAdapter(backendCatalog));
+
+    RESTCatalog catalog =
+        new RESTCatalog(SessionCatalog.SessionContext.createEmpty(), (config) -> adapter);
+    catalog.initialize(
+        "test",
+        ImmutableMap.of(
+            CatalogProperties.FILE_IO_IMPL, "org.apache.iceberg.inmemory.InMemoryFileIO"));
+
+    if (requiresNamespaceCreate()) {
+      catalog.createNamespace(TABLE.namespace());
+    }
+
+    catalog.createTable(TABLE, SCHEMA);
+
+    // Simulate: server returns CommitStateUnknown and does NOT apply the commit
+    Mockito.doThrow(
+            new CommitStateUnknownException(new ServiceFailureException("Service failed: 503")))
+        .when(adapter)
+        .execute(
+            reqMatcher(HTTPMethod.POST, RESOURCE_PATHS.table(TABLE)),
+            eq(LoadTableResponse.class),
+            any(),
+            any());
+
+    Table table = catalog.loadTable(TABLE);
+
+    assertThatThrownBy(() -> table.newFastAppend().appendFile(FILE_A).commit())
+        .isInstanceOf(CommitStateUnknownException.class)
+        .hasMessageContaining("Cannot determine whether the commit was successful")
+        .satisfies(ex -> assertThat(((CommitStateUnknownException) ex).getSuppressed()).isEmpty());
+  }
+
   private RESTCatalog catalog(RESTCatalogAdapter adapter) {
     RESTCatalog catalog =
         new RESTCatalog(SessionCatalog.SessionContext.createEmpty(), (config) -> adapter);