Add retry logic for notification updates on concurrent entity modification

CHANGELOG.md

@@ -59,6 +59,7 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti
 - Fixed JDBC persistence under `SERIALIZABLE` isolation (e.g. CockroachDB default) so that a concurrent entity create that loses a unique-name race no longer returns the phantom new entity as a successful create. The conflicting row is now reported as `ENTITY_ALREADY_EXISTS` instead of fabricating the entity that was not persisted.
 - Python CLI `setup` now preserves `endpoint_internal` and `sts_endpoint` during apply and export for S3 configuration
 - Fixed a false-negative in the JDBC optimized location-overlap check (`OPTIMIZED_SIBLING_CHECK`). Ancestor locations stored in `location_without_scheme` without a trailing slash were not matched by the generated ancestor equality terms, allowing nested table/namespace locations to be created under existing prefixes. The query now emits both slash-terminated and non-slash-terminated prefix terms and uses a slash-terminated `LIKE` pattern for descendant matching.
+- Notification UPDATE requests for external tables now retry on concurrent entity modifications instead of failing immediately. The retry re-reads the latest entity from the metastore, re-validates the notification timestamp, and retries the update up to 3 times before giving up.
 
 ### Commits
 

runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/LocalIcebergCatalog.java

@@ -3103,23 +3103,74 @@ private boolean sendNotificationForTableLike(
       // finally, validate that the metadata file is within the table directory
       validateMetadataFileInTableDir(tableIdentifier, tableMetadata);
 
-      // TODO: These might fail due to concurrent update; we need to do a retry in those cases.
       if (null == existingLocation) {
         LOGGER.debug(
             "Creating table {} for notification with metadataLocation {}",
             tableIdentifier,
             newLocation);
         createTableLike(tableIdentifier, entity, resolvedParent, false);
       } else {
+        updateTableLikeForNotification(
+            tableIdentifier, entity, newLocation, request.getPayload().getTimestamp());
+      }
+    }
+    return true;
+  }
+
+  private static final int MAX_NOTIFICATION_UPDATE_ATTEMPTS = 3;
+
+  private void updateTableLikeForNotification(
+      TableIdentifier tableIdentifier,
+      IcebergTableLikeEntity entity,
+      String newLocation,
+      long notificationTimestamp) {
+    for (int attempt = 1; ; attempt++) {
+      try {
         LOGGER.debug(
-            "Updating table {} for notification with metadataLocation {}",
+            "Updating table {} for notification with metadataLocation {} (attempt {}/{})",
             tableIdentifier,
-            newLocation);
-
+            newLocation,
+            attempt,
+            MAX_NOTIFICATION_UPDATE_ATTEMPTS);
         updateTableLike(tableIdentifier, entity, false);
+        return;
+      } catch (CommitConflictException e) {
+        if (attempt >= MAX_NOTIFICATION_UPDATE_ATTEMPTS) {
+          throw e;
+        }
+        LOGGER
+            .atInfo()
+            .addKeyValue(StructuredLogKeys.TABLE_IDENTIFIER, tableIdentifier)
+            .addKeyValue("attempt", attempt)
+            .log("Concurrent modification during notification update, retrying");
+
+        EntityResult reloadResult =
+            getMetaStoreManager()
+                .loadEntity(
+                    getCurrentPolarisContext(),
+                    entity.getCatalogId(),
+                    entity.getId(),
+                    PolarisEntityType.TABLE_LIKE);
+        if (!reloadResult.isSuccess()) {
+          throw e;
+        }
+        IcebergTableLikeEntity freshEntity = IcebergTableLikeEntity.of(reloadResult.getEntity());
+        if (freshEntity == null) {
+          throw e;
+        }
+        if (freshEntity.getLastAdmittedNotificationTimestamp().isPresent()
+            && notificationTimestamp <= freshEntity.getLastAdmittedNotificationTimestamp().get()) {
+          throw new AlreadyExistsException(
+              "A notification with a newer timestamp has been processed for table %s",
+              tableIdentifier);
+        }
+        entity =
+            new IcebergTableLikeEntity.Builder(freshEntity)
+                .setMetadataLocation(newLocation)
+                .setLastNotificationTimestamp(notificationTimestamp)
+                .build();
       }
     }
-    return true;
   }
 
   private void createNonExistingNamespaces(Namespace namespace) {

runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/AbstractLocalIcebergCatalogTest.java

@@ -60,6 +60,7 @@
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -1217,6 +1218,137 @@ public void testUpdateNotificationWhenTableAndNamespacesDontExistNamespaceRaceCo
         .isTrue();
   }
 
+  @Test
+  public void testNotificationUpdateRetriesOnConcurrentModification() {
+    Assumptions.assumeTrue(
+        requiresNamespaceCreate(),
+        "Only applicable if namespaces must be created before adding children");
+    Assumptions.assumeTrue(
+        supportsNestedNamespaces(), "Only applicable if nested namespaces are supported");
+    Assumptions.assumeTrue(
+        supportsNotifications(), "Only applicable if notifications are supported");
+
+    final String tableLocation = "s3://externally-owned-bucket/retry-table/";
+    final String createMetadataLocation = tableLocation + "metadata/v1.metadata.json";
+    final String updateMetadataLocation = tableLocation + "metadata/v2.metadata.json";
+
+    PolarisMetaStoreManager spyMetaStore = spy(metaStoreManager);
+    LocalIcebergCatalog catalog = newIcebergCatalog(CATALOG_NAME, spyMetaStore);
+    catalog.initialize(
+        CATALOG_NAME,
+        ImmutableMap.of(
+            CatalogProperties.FILE_IO_IMPL, "org.apache.iceberg.inmemory.InMemoryFileIO"));
+
+    Namespace namespace = Namespace.of("parent", "child1");
+    TableIdentifier table = TableIdentifier.of(namespace, "retry_table");
+
+    fileIO.addFile(
+        createMetadataLocation,
+        TableMetadataParser.toJson(createSampleTableMetadata(tableLocation)).getBytes(UTF_8));
+    fileIO.addFile(
+        updateMetadataLocation,
+        TableMetadataParser.toJson(createSampleTableMetadata(tableLocation)).getBytes(UTF_8));
+
+    NotificationRequest createRequest = new NotificationRequest();
+    createRequest.setNotificationType(NotificationType.CREATE);
+    TableUpdateNotification createPayload = new TableUpdateNotification();
+    createPayload.setMetadataLocation(createMetadataLocation);
+    createPayload.setTableName(table.name());
+    createPayload.setTableUuid(UUID.randomUUID().toString());
+    createPayload.setTimestamp(100L);
+    createRequest.setPayload(createPayload);
+
+    catalog.sendNotification(table, createRequest);
+    Assertions.assertThat(catalog.tableExists(table)).isTrue();
+
+    AtomicInteger updateAttempts = new AtomicInteger();
+    doAnswer(
+            invocation -> {
+              if (updateAttempts.incrementAndGet() == 1) {
+                return new EntityResult(
+                    BaseResult.ReturnStatus.TARGET_ENTITY_CONCURRENTLY_MODIFIED, null);
+              }
+              return invocation.callRealMethod();
+            })
+        .when(spyMetaStore)
+        .updateEntityPropertiesIfNotChanged(any(), any(), any());
+
+    NotificationRequest updateRequest = new NotificationRequest();
+    updateRequest.setNotificationType(NotificationType.UPDATE);
+    TableUpdateNotification updatePayload = new TableUpdateNotification();
+    updatePayload.setMetadataLocation(updateMetadataLocation);
+    updatePayload.setTableName(table.name());
+    updatePayload.setTableUuid(UUID.randomUUID().toString());
+    updatePayload.setTimestamp(200L);
+    updateRequest.setPayload(updatePayload);
+
+    Assertions.assertThat(catalog.sendNotification(table, updateRequest))
+        .as("Notification should succeed after retry")
+        .isTrue();
+    Assertions.assertThat(updateAttempts.get())
+        .as("Should have retried once after concurrent modification")
+        .isEqualTo(2);
+  }
+
+  @Test
+  public void testNotificationUpdateGivesUpAfterMaxRetries() {
+    Assumptions.assumeTrue(
+        requiresNamespaceCreate(),
+        "Only applicable if namespaces must be created before adding children");
+    Assumptions.assumeTrue(
+        supportsNestedNamespaces(), "Only applicable if nested namespaces are supported");
+    Assumptions.assumeTrue(
+        supportsNotifications(), "Only applicable if notifications are supported");
+
+    final String tableLocation = "s3://externally-owned-bucket/exhaust-table/";
+    final String createMetadataLocation = tableLocation + "metadata/v1.metadata.json";
+    final String updateMetadataLocation = tableLocation + "metadata/v2.metadata.json";
+
+    PolarisMetaStoreManager spyMetaStore = spy(metaStoreManager);
+    LocalIcebergCatalog catalog = newIcebergCatalog(CATALOG_NAME, spyMetaStore);
+    catalog.initialize(
+        CATALOG_NAME,
+        ImmutableMap.of(
+            CatalogProperties.FILE_IO_IMPL, "org.apache.iceberg.inmemory.InMemoryFileIO"));
+
+    Namespace namespace = Namespace.of("parent", "child1");
+    TableIdentifier table = TableIdentifier.of(namespace, "exhaust_table");
+
+    fileIO.addFile(
+        createMetadataLocation,
+        TableMetadataParser.toJson(createSampleTableMetadata(tableLocation)).getBytes(UTF_8));
+    fileIO.addFile(
+        updateMetadataLocation,
+        TableMetadataParser.toJson(createSampleTableMetadata(tableLocation)).getBytes(UTF_8));
+
+    NotificationRequest createRequest = new NotificationRequest();
+    createRequest.setNotificationType(NotificationType.CREATE);
+    TableUpdateNotification createPayload = new TableUpdateNotification();
+    createPayload.setMetadataLocation(createMetadataLocation);
+    createPayload.setTableName(table.name());
+    createPayload.setTableUuid(UUID.randomUUID().toString());
+    createPayload.setTimestamp(100L);
+    createRequest.setPayload(createPayload);
+
+    catalog.sendNotification(table, createRequest);
+
+    doReturn(new EntityResult(BaseResult.ReturnStatus.TARGET_ENTITY_CONCURRENTLY_MODIFIED, null))
+        .when(spyMetaStore)
+        .updateEntityPropertiesIfNotChanged(any(), any(), any());
+
+    NotificationRequest updateRequest = new NotificationRequest();
+    updateRequest.setNotificationType(NotificationType.UPDATE);
+    TableUpdateNotification updatePayload = new TableUpdateNotification();
+    updatePayload.setMetadataLocation(updateMetadataLocation);
+    updatePayload.setTableName(table.name());
+    updatePayload.setTableUuid(UUID.randomUUID().toString());
+    updatePayload.setTimestamp(200L);
+    updateRequest.setPayload(updatePayload);
+
+    Assertions.assertThatThrownBy(() -> catalog.sendNotification(table, updateRequest))
+        .isInstanceOf(CommitConflictException.class);
+  }
+
   @Test
   public void testUpdateNotificationCreateTableInDisallowedLocation() {
     Assumptions.assumeTrue(