Fix stale snapshot detection to return 409 instead of 400 (#425)

## Summary
- Fix stale snapshot detection during concurrent modifications to return
HTTP 409 (Conflict) instead of HTTP 400 (Bad Request)
- Reclassify `ValidationException` with stale snapshot message to
`CommitFailedException` (409) to allow client retry
- Ensure other ValidationException instances are handled as HTTP 400 Bad
Request responses (e.g., attempting to delete a non-existent snapshot)

## Problem
When concurrent modifications occur during a transaction commit:
1. Client builds snapshots based on table version N (e.g.,
`lastSequenceNumber = 4`)
2. Client sends commit request with these snapshots in
`SNAPSHOTS_JSON_KEY`
3. Meanwhile, another process commits version N+1 (e.g.,
`lastSequenceNumber = 5`)
4. Server calls `doRefresh()` which updates `current()` to version N+1
5. **Bug:** The snapshots in `SNAPSHOTS_JSON_KEY` are now stale (their
sequence numbers are based on version N)
6. Iceberg's `TableMetadata.addSnapshot()` throws `ValidationException`
→ mapped to 400 Bad Request
7. Should return 409 Conflict so clients know to refresh and retry

## Solution
Let Iceberg's existing validation detect sequence number conflicts, then
catch the `ValidationException` and reclassify it as
`CommitFailedException` for the specific stale snapshot error pattern:

```java
} catch (ValidationException e) {
  // Stale snapshot errors are retryable - client should refresh and retry
  if (isStaleSnapshotError(e)) {
    throw new CommitFailedException(e);
  }
  throw new BadRequestException(e, e.getMessage());
}
```

This approach is simpler than pre-checking and leverages Iceberg's
existing validation.

## Test Plan
- [x] Unit test `testStaleSnapshotErrorDetection()` verifies error
detection logic
- [x] All existing internalcatalog tests pass
- [ ] Integration testing in staging environment

---------

Co-authored-by: Claude Opus 4.5 <[email protected]>

Author: cbb330

iceberg/openhouse/internalcatalog/src/main/java/com/linkedin/openhouse/internal/catalog/OpenHouseInternalTableOperations.java

@@ -383,7 +383,13 @@ updatedMtDataRef, io().newOutputFile(newMetadataLocation)),
             e);
       }
       throw new CommitFailedException(ioe);
-    } catch (InvalidIcebergSnapshotException | IllegalArgumentException | ValidationException e) {
+    } catch (InvalidIcebergSnapshotException | IllegalArgumentException e) {
+      throw new BadRequestException(e, e.getMessage());
+    } catch (ValidationException e) {
+      // Stale snapshot errors are retryable - client should refresh and retry
+      if (isStaleSnapshotError(e)) {
+        throw new CommitFailedException(e);
+      }
       throw new BadRequestException(e, e.getMessage());
     } catch (CommitFailedException e) {
       throw e;
@@ -556,6 +562,17 @@ private void failIfRetryUpdate(Map<String, String> properties) {
     }
   }
 
+  /**
+   * Checks if a ValidationException is due to a stale snapshot (sequence number conflict). This
+   * happens during concurrent modifications and should be retryable (409), not a bad request (400).
+   */
+  private boolean isStaleSnapshotError(ValidationException e) {
+    String msg = e.getMessage();
+    return msg != null
+        && msg.contains("Cannot add snapshot with sequence number")
+        && msg.contains("older than last sequence number");
+  }
+
   /**
    * Process all schema-related operations including client schema (for new tables), evolved schema
    * (for updates), and intermediate schemas (for replication scenarios). This consolidates all

iceberg/openhouse/internalcatalog/src/test/java/com/linkedin/openhouse/internal/catalog/IcebergTestUtil.java

@@ -16,6 +16,7 @@ public final class IcebergTestUtil {
   private static final String SNAPSHOTS_FILE = "serialized_snapshots.json";
   private static final String EXTRA_SNAPSHOTS_FILE = "extra_serialized_snapshots.json";
   private static final String WAP_SNAPSHOTS_FILE = "wap_serialized_snapshots.json";
+  private static final String STALE_SNAPSHOT_FILE = "stale_snapshot.json";
 
   private IcebergTestUtil() {}
 
@@ -31,6 +32,10 @@ public static List<Snapshot> getWapSnapshots() throws IOException {
     return loadSnapshots(WAP_SNAPSHOTS_FILE);
   }
 
+  public static List<Snapshot> getStaleSnapshots() throws IOException {
+    return loadSnapshots(STALE_SNAPSHOT_FILE);
+  }
+
   private static List<Snapshot> loadSnapshots(String snapshotFile) throws IOException {
     InputStream inputStream =
         IcebergTestUtil.class.getClassLoader().getResourceAsStream(snapshotFile);

iceberg/openhouse/internalcatalog/src/test/java/com/linkedin/openhouse/internal/catalog/OpenHouseInternalTableOperationsTest.java

@@ -52,6 +52,7 @@
 import org.apache.iceberg.exceptions.BadRequestException;
 import org.apache.iceberg.exceptions.CommitFailedException;
 import org.apache.iceberg.exceptions.CommitStateUnknownException;
+import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.hadoop.HadoopFileIO;
 import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
@@ -77,7 +78,7 @@ public class OpenHouseInternalTableOperationsTest {
               Types.NestedField.required(2, "ts", Types.TimestampType.withoutZone())),
           PartitionSpec.unpartitioned(),
           getTempLocation(),
-          ImmutableMap.of());
+          ImmutableMap.of("format-version", "2"));
   @Mock private HouseTableRepository mockHouseTableRepository;
   @Mock private HouseTableMapper mockHouseTableMapper;
   @Mock private HouseTable mockHouseTable;
@@ -1718,4 +1719,100 @@ void testMultipleDiffCommitWithWAPSnapshots() throws IOException {
       Mockito.verify(mockHouseTableRepository, Mockito.times(1)).save(Mockito.eq(mockHouseTable));
     }
   }
+
+  /**
+   * Tests that stale snapshot detection returns 409 Conflict instead of 400 Bad Request.
+   *
+   * <p>This test reproduces the production scenario where:
+   *
+   * <ol>
+   *   <li>Table has 4 existing snapshots with lastSequenceNumber = 4
+   *   <li>A concurrent modification occurs during commit
+   *   <li>Client tries to add a snapshot with sequenceNumber = 4 (now stale)
+   *   <li>Before fix: ValidationException → BadRequestException (400)
+   *   <li>After fix: Stale snapshot detected → CommitFailedException (409)
+   * </ol>
+   *
+   * <p>The 409 response tells clients to refresh and retry, while 400 incorrectly suggests the
+   * request was invalid.
+   */
+  /**
+   * First verifies that Iceberg's validation actually throws for stale snapshots in format v2, then
+   * tests the OpenHouse doCommit path.
+   */
+  @Test
+  void testStaleSnapshotDuringConcurrentModificationReturns409NotBadRequest() throws IOException {
+    // Build base metadata with all 4 test snapshots (lastSequenceNumber = 4)
+    List<Snapshot> existingSnapshots = IcebergTestUtil.getSnapshots();
+    TableMetadata tempMetadata = BASE_TABLE_METADATA;
+    for (Snapshot snapshot : existingSnapshots) {
+      tempMetadata =
+          TableMetadata.buildFrom(tempMetadata)
+              .setBranchSnapshot(snapshot, SnapshotRef.MAIN_BRANCH)
+              .build();
+    }
+    final TableMetadata baseMetadata = tempMetadata;
+
+    // Verify base metadata has format version 2 and lastSequenceNumber = 4
+    Assertions.assertEquals(2, baseMetadata.formatVersion(), "Format version should be 2");
+    Assertions.assertEquals(4, baseMetadata.lastSequenceNumber(), "lastSequenceNumber should be 4");
+
+    // Load stale snapshot with sequenceNumber = 4 (same as lastSequenceNumber)
+    List<Snapshot> staleSnapshots = IcebergTestUtil.getStaleSnapshots();
+    Snapshot staleSnapshot = staleSnapshots.get(0);
+    Assertions.assertEquals(
+        4, staleSnapshot.sequenceNumber(), "Stale snapshot should have sequenceNumber = 4");
+
+    // Verify stale snapshot has a parent (required for Iceberg 1.5+ validation)
+    Assertions.assertNotNull(
+        staleSnapshot.parentId(), "Stale snapshot must have parentId for validation to trigger");
+
+    // FIRST: Verify Iceberg's validation works directly
+    TableMetadata.Builder directBuilder = TableMetadata.buildFrom(baseMetadata);
+    ValidationException icebergException =
+        Assertions.assertThrows(
+            ValidationException.class,
+            () -> directBuilder.addSnapshot(staleSnapshot),
+            "Iceberg should throw ValidationException for stale snapshot");
+    Assertions.assertTrue(
+        icebergException.getMessage().contains("Cannot add snapshot with sequence number"),
+        "Iceberg should report sequence number issue: " + icebergException.getMessage());
+
+    // NOW test the full doCommit path
+    List<Snapshot> allSnapshots = new ArrayList<>(existingSnapshots);
+    allSnapshots.addAll(staleSnapshots);
+
+    Map<String, String> properties = new HashMap<>(baseMetadata.properties());
+    properties.put(
+        CatalogConstants.SNAPSHOTS_JSON_KEY, SnapshotsUtil.serializedSnapshots(allSnapshots));
+    properties.put(
+        CatalogConstants.SNAPSHOTS_REFS_KEY,
+        SnapshotsUtil.serializeMap(
+            IcebergTestUtil.createMainBranchRefPointingTo(existingSnapshots.get(3))));
+    properties.put(getCanonicalFieldName("tableLocation"), TEST_LOCATION);
+
+    final TableMetadata metadataWithStaleSnapshot = baseMetadata.replaceProperties(properties);
+
+    try (MockedStatic<TableMetadataParser> ignoreWriteMock =
+        Mockito.mockStatic(TableMetadataParser.class)) {
+
+      // Should throw CommitFailedException (409), not BadRequestException (400)
+      Exception exception =
+          Assertions.assertThrows(
+              CommitFailedException.class,
+              () ->
+                  openHouseInternalTableOperations.doCommit(
+                      baseMetadata, metadataWithStaleSnapshot),
+              "Stale snapshot should return 409 Conflict (CommitFailedException), not 400 "
+                  + "(BadRequestException). 400 suggests invalid request, 409 suggests retry.");
+
+      String message = exception.getMessage().toLowerCase();
+      Assertions.assertTrue(
+          message.contains("stale")
+              || message.contains("sequence")
+              || message.contains("concurrent"),
+          "Exception message should indicate stale snapshot or sequence number issue: "
+              + exception.getMessage());
+    }
+  }
 }

iceberg/openhouse/internalcatalog/src/test/resources/stale_snapshot.json

@@ -0,0 +1,3 @@
+[
+  "{  \"snapshot-id\" : 9999888777666555,  \"parent-snapshot-id\" : 4151407017102313398,  \"sequence-number\" : 4,  \"timestamp-ms\" : 1767258030812,  \"summary\" : {    \"operation\" : \"append\",    \"spark.app.id\" : \"local-1767258030000\",    \"added-data-files\" : \"1\",    \"added-records\" : \"1\",    \"added-files-size\" : \"673\",    \"changed-partition-count\" : \"1\",    \"total-records\" : \"1\",    \"total-files-size\" : \"673\",    \"total-data-files\" : \"1\",    \"total-delete-files\" : \"0\",    \"total-position-deletes\" : \"0\",    \"total-equality-deletes\" : \"0\"  },  \"manifest-list\" : \"/data/openhouse/tracking_live_daily/SponsoredDocumentPageImpressionEvent/metadata/snap-9999888777666555-1-aa0dcbb9-707f-4f53-9df8-394bad8563f2.avro\",  \"schema-id\" : 0}"
+]