Add Events for Iceberg REST APIs (#2480)

Author: adnanhemani

runtime/service/src/main/java/org/apache/polaris/service/admin/PolarisServiceImpl.java

@@ -76,7 +76,6 @@
 import org.apache.polaris.service.admin.api.PolarisPrincipalRolesApiService;
 import org.apache.polaris.service.admin.api.PolarisPrincipalsApiService;
 import org.apache.polaris.service.config.ReservedProperties;
-import org.apache.polaris.service.events.AfterCatalogCreatedEvent;
 import org.apache.polaris.service.events.listeners.PolarisEventListener;
 import org.apache.polaris.service.types.PolicyIdentifier;
 import org.slf4j.Logger;
@@ -131,7 +130,6 @@ public Response createCatalog(
     validateExternalCatalog(catalog);
     Catalog newCatalog = CatalogEntity.of(adminService.createCatalog(request)).asCatalog();
     LOGGER.info("Created new catalog {}", newCatalog);
-    polarisEventListener.onAfterCatalogCreated(new AfterCatalogCreatedEvent(newCatalog.getName()));
     return Response.status(Response.Status.CREATED).build();
   }
 

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

@@ -132,14 +132,7 @@
 import org.apache.polaris.service.catalog.io.FileIOFactory;
 import org.apache.polaris.service.catalog.io.FileIOUtil;
 import org.apache.polaris.service.catalog.validation.IcebergPropertiesValidation;
-import org.apache.polaris.service.events.AfterTableCommitedEvent;
-import org.apache.polaris.service.events.AfterTableRefreshedEvent;
-import org.apache.polaris.service.events.AfterViewCommitedEvent;
-import org.apache.polaris.service.events.AfterViewRefreshedEvent;
-import org.apache.polaris.service.events.BeforeTableCommitedEvent;
-import org.apache.polaris.service.events.BeforeTableRefreshedEvent;
-import org.apache.polaris.service.events.BeforeViewCommitedEvent;
-import org.apache.polaris.service.events.BeforeViewRefreshedEvent;
+import org.apache.polaris.service.events.IcebergRestCatalogEvents;
 import org.apache.polaris.service.events.listeners.PolarisEventListener;
 import org.apache.polaris.service.task.TaskExecutor;
 import org.apache.polaris.service.types.NotificationRequest;
@@ -1446,8 +1439,8 @@ public void doRefresh() {
       if (latestLocation == null) {
         disableRefresh();
       } else {
-        polarisEventListener.onBeforeTableRefreshed(
-            new BeforeTableRefreshedEvent(catalogName, tableIdentifier));
+        polarisEventListener.onBeforeRefreshTable(
+            new IcebergRestCatalogEvents.BeforeRefreshTableEvent(catalogName, tableIdentifier));
         refreshFromMetadataLocation(
             latestLocation,
             SHOULD_RETRY_REFRESH_PREDICATE,
@@ -1467,14 +1460,15 @@ public void doRefresh() {
                       Set.of(PolarisStorageActions.READ, PolarisStorageActions.LIST));
               return TableMetadataParser.read(fileIO, metadataLocation);
             });
-        polarisEventListener.onAfterTableRefreshed(
-            new AfterTableRefreshedEvent(catalogName, tableIdentifier));
+        polarisEventListener.onAfterRefreshTable(
+            new IcebergRestCatalogEvents.AfterRefreshTableEvent(catalogName, tableIdentifier));
       }
     }
 
     public void doCommit(TableMetadata base, TableMetadata metadata) {
-      polarisEventListener.onBeforeTableCommited(
-          new BeforeTableCommitedEvent(tableIdentifier, base, metadata));
+      polarisEventListener.onBeforeCommitTable(
+          new IcebergRestCatalogEvents.BeforeCommitTableEvent(
+              catalogName, tableIdentifier, base, metadata));
 
       LOGGER.debug(
           "doCommit for table {} with metadataBefore {}, metadataAfter {}",
@@ -1618,8 +1612,9 @@ public void doCommit(TableMetadata base, TableMetadata metadata) {
         updateTableLike(tableIdentifier, entity);
       }
 
-      polarisEventListener.onAfterTableCommited(
-          new AfterTableCommitedEvent(catalogName, tableIdentifier, base, metadata));
+      polarisEventListener.onAfterCommitTable(
+          new IcebergRestCatalogEvents.AfterCommitTableEvent(
+              catalogName, tableIdentifier, base, metadata));
     }
 
     @Override
@@ -1810,8 +1805,8 @@ public void doRefresh() {
       if (latestLocation == null) {
         disableRefresh();
       } else {
-        polarisEventListener.onBeforeViewRefreshed(
-            new BeforeViewRefreshedEvent(catalogName, identifier));
+        polarisEventListener.onBeforeRefreshView(
+            new IcebergRestCatalogEvents.BeforeRefreshViewEvent(catalogName, identifier));
         refreshFromMetadataLocation(
             latestLocation,
             SHOULD_RETRY_REFRESH_PREDICATE,
@@ -1833,14 +1828,15 @@ public void doRefresh() {
 
               return ViewMetadataParser.read(fileIO.newInputFile(metadataLocation));
             });
-        polarisEventListener.onAfterViewRefreshed(
-            new AfterViewRefreshedEvent(catalogName, identifier));
+        polarisEventListener.onAfterRefreshView(
+            new IcebergRestCatalogEvents.AfterRefreshViewEvent(catalogName, identifier));
       }
     }
 
     public void doCommit(ViewMetadata base, ViewMetadata metadata) {
-      polarisEventListener.onBeforeViewCommited(
-          new BeforeViewCommitedEvent(catalogName, identifier, base, metadata));
+      polarisEventListener.onBeforeCommitView(
+          new IcebergRestCatalogEvents.BeforeCommitViewEvent(
+              catalogName, identifier, base, metadata));
 
       // TODO: Maybe avoid writing metadata if there's definitely a transaction conflict
       LOGGER.debug(
@@ -1940,8 +1936,9 @@ public void doCommit(ViewMetadata base, ViewMetadata metadata) {
         updateTableLike(identifier, entity);
       }
 
-      polarisEventListener.onAfterViewCommited(
-          new AfterViewCommitedEvent(catalogName, identifier, base, metadata));
+      polarisEventListener.onAfterCommitView(
+          new IcebergRestCatalogEvents.AfterCommitViewEvent(
+              catalogName, identifier, base, metadata));
     }
 
     protected String writeNewMetadataIfRequired(ViewMetadata metadata) {

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

@@ -102,7 +102,6 @@
 import org.apache.polaris.service.catalog.common.CatalogHandler;
 import org.apache.polaris.service.config.ReservedProperties;
 import org.apache.polaris.service.context.catalog.CallContextCatalogFactory;
-import org.apache.polaris.service.events.AfterTableCreatedEvent;
 import org.apache.polaris.service.events.listeners.PolarisEventListener;
 import org.apache.polaris.service.http.IcebergHttpUtil;
 import org.apache.polaris.service.http.IfNoneMatch;
@@ -392,11 +391,8 @@ public LoadTableResponse createTableDirect(Namespace namespace, CreateTableReque
             .withWriteOrder(request.writeOrder())
             .setProperties(reservedProperties.removeReservedProperties(request.properties()))
             .build();
-    LoadTableResponse resp =
-        catalogHandlerUtils.createTable(baseCatalog, namespace, requestWithoutReservedProperties);
-    polarisEventListener.onAfterTableCreated(
-        new AfterTableCreatedEvent(catalogName, identifier, resp.tableMetadata()));
-    return resp;
+    return catalogHandlerUtils.createTable(
+        baseCatalog, namespace, requestWithoutReservedProperties);
   }
 
   /**