diff --git a/CHANGELOG.md b/CHANGELOG.md index 38abee8da9..a2f354730a 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/LocalIcebergCatalog.java b/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/LocalIcebergCatalog.java index 1cea9890c3..3199123c8a 100644 --- a/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/LocalIcebergCatalog.java +++ b/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/LocalIcebergCatalog.java @@ -3103,7 +3103,6 @@ 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 {}", @@ -3111,15 +3110,67 @@ private boolean sendNotificationForTableLike( 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) { diff --git a/runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/AbstractLocalIcebergCatalogTest.java b/runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/AbstractLocalIcebergCatalogTest.java index 8da58c833c..9531bcc2b1 100644 --- a/runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/AbstractLocalIcebergCatalogTest.java +++ b/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(