@Test
  public void testSequenceNumberConflictThrowsCommitFailed() {
    RESTCatalogAdapter adapter = Mockito.spy(new RESTCatalogAdapter(backendCatalog));
    RESTCatalog catalog = catalog(adapter);

    catalog.createNamespace(TABLE.namespace());
    catalog.buildTable(TABLE, SCHEMA).withPartitionSpec(SPEC).create();

    DataFile fileOnMain =
        DataFiles.builder(SPEC)
            .withPath("/path/commit-test-file1.parquet")
            .withFileSizeInBytes(10)
            .withPartitionPath("id_bucket=0")
            .withRecordCount(1)
            .build();

    catalog.loadTable(TABLE).newFastAppend().appendFile(fileOnMain).commit();

    DataFile fileOnAnotherBranch =
        DataFiles.builder(SPEC)
            .withPath("/path/commit-test-conflicting.parquet")
            .withFileSizeInBytes(10)
            .withPartitionPath("id_bucket=0")
            .withRecordCount(1)
            .build();

    // Before the next commit is processed by the server, advance the server's lastSequenceNumber
    // by committing to a different branch. This simulates a concurrent request to a different branch
    // that "beats" the commit to main.
    Mockito.doAnswer(
            invocation -> {
              backendCatalog
                  .loadTable(TABLE)
                  .newFastAppend()
                  .appendFile(fileOnAnotherBranch)
                  .toBranch("other")
                  .commit();
              return invocation.callRealMethod();
            })
        .when(adapter)
        .execute(matches(HTTPMethod.POST, RESOURCE_PATHS.table(TABLE)), any(), any(), any());

    DataFile anotherFileOnMain =
        DataFiles.builder(SPEC)
            .withPath("/path/commit-test-file2.parquet")
            .withFileSizeInBytes(10)
            .withPartitionPath("id_bucket=0")
            .withRecordCount(1)
            .build();

    assertThatThrownBy(
            () -> catalog.loadTable(TABLE).newFastAppend().appendFile(anotherFileOnMain).commit())
        .isInstanceOf(CommitFailedException.class)
        .hasMessageContaining("Validation failed, please retry");
  }

  @Test
  public void testAddSnapshotWithStaleSequenceNumberIsRetryable() {
    TableMetadata base = TableMetadata.newTableMetadata(
            TEST_SCHEMA, PartitionSpec.unpartitioned(), "location", ImmutableMap.of());

    // Advance lastSequenceNumber to 1 by adding a root snapshot
    Snapshot s1 = new BaseSnapshot(
            1, 1L, null, System.currentTimeMillis(), null, null, null, "file:/s1.avro", null, null, null);
    TableMetadata withS1 = TableMetadata.buildFrom(base).addSnapshot(s1).build();

    // A snapshot with seqNum=1 and non-null parentId is stale (1 is not > lastSequenceNumber=1)
    Snapshot staleSnapshot = new BaseSnapshot(
            1, 2L, 1L, System.currentTimeMillis(), null, null, null, "file:/s2.avro", null, null, null);

    assertThatThrownBy(() -> TableMetadata.buildFrom(withS1).addSnapshot(staleSnapshot))
            .isInstanceOf(RetryableValidationException.class)
            .hasMessageContaining("Cannot add snapshot with sequence number");
  }

  @Test
  public void testAddSnapshotWithStaleFirstRowIdIsRetryable() {
    TableMetadata base = TableMetadata.newTableMetadata(
            TEST_SCHEMA, PartitionSpec.unpartitioned(), "location",
            ImmutableMap.of(TableProperties.FORMAT_VERSION, "3"));

    // Advance nextRowId to 5 by adding a snapshot that allocates 5 rows
    Snapshot s1 = new BaseSnapshot(
            1, 1L, null, System.currentTimeMillis(), null, null, null, "file:/s1.avro", 0L, 5L, null);
    TableMetadata withS1 = TableMetadata.buildFrom(base).addSnapshot(s1).build();
    // nextRowId is now 5

    // A snapshot with firstRowId=0 is stale (0 < nextRowId=5)
    Snapshot staleSnapshot = new BaseSnapshot(
            2, 2L, 1L, System.currentTimeMillis(), null, null, null, "file:/s2.avro", 0L, 1L, null);

    assertThatThrownBy(() -> TableMetadata.buildFrom(withS1).addSnapshot(staleSnapshot))
            .isInstanceOf(RetryableValidationException.class)
            .hasMessageContaining("Cannot add a snapshot, first-row-id is behind table next-row-id");
  }