Enable multiple event listeners to consume for Polaris events

CHANGELOG.md

@@ -79,6 +79,7 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti
 
 - The configuration option `polaris.rate-limiter.token-bucket.window` is no longer supported and should be removed.
 - `PolarisConfigurationStore` has been deprecated for removal.
+- The configuration option `polaris.event-listener.type` is deprecated and will be removed later. Please use `polaris.event-listener.types` instead.
 
 ### Fixes
 

extensions/auth/opa/tests/src/intTest/java/org/apache/polaris/extension/auth/opa/test/OpaTestProfiles.java

@@ -19,7 +19,6 @@
 package org.apache.polaris.extension.auth.opa.test;
 
 import io.quarkus.test.junit.QuarkusTestProfile;
-import io.quarkus.test.junit.QuarkusTestProfile.TestResourceEntry;
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;

polaris-core/src/main/java/org/apache/polaris/core/connection/OAuthClientCredentialsParametersDpo.java

@@ -20,7 +20,6 @@
 
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
-import com.google.common.base.Joiner;
 import com.google.common.base.MoreObjects;
 import jakarta.annotation.Nonnull;
 import jakarta.annotation.Nullable;
@@ -43,9 +42,6 @@
  * API model.
  */
 public class OAuthClientCredentialsParametersDpo extends AuthenticationParametersDpo {
-
-  private static final Joiner COLON_JOINER = Joiner.on(":");
-
   @JsonProperty(value = "tokenUri")
   private final String tokenUri;
 

runtime/defaults/src/main/resources/application.properties

@@ -157,8 +157,6 @@ quarkus.rds.sync-client.type=apache
 
 polaris.file-io.type=default
 
-polaris.event-listener.type=no-op
-
 # Persistence event listener settings
 # polaris.event-listener.type=persistence-in-memory-buffer
 # polaris.event-listener.persistence-in-memory-buffer.buffer-time=5000ms

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

@@ -45,29 +45,29 @@
 import org.apache.polaris.service.events.EventAttributeMap;
 import org.apache.polaris.service.events.EventAttributes;
 import org.apache.polaris.service.events.PolarisEvent;
+import org.apache.polaris.service.events.PolarisEventDispatcher;
 import org.apache.polaris.service.events.PolarisEventMetadataFactory;
 import org.apache.polaris.service.events.PolarisEventType;
-import org.apache.polaris.service.events.listeners.PolarisEventListener;
 
 @Decorator
 @Priority(1000)
 public class PolarisCatalogsEventServiceDelegator implements PolarisCatalogsApiService {
 
   @Inject @Delegate PolarisCatalogsApiService delegate;
-  @Inject PolarisEventListener polarisEventListener;
+  @Inject PolarisEventDispatcher polarisEventDispatcher;
   @Inject PolarisEventMetadataFactory eventMetadataFactory;
 
   @Override
   public Response createCatalog(
       CreateCatalogRequest request, RealmContext realmContext, SecurityContext securityContext) {
-    polarisEventListener.onEvent(
+    polarisEventDispatcher.dispatch(
         new PolarisEvent(
             PolarisEventType.BEFORE_CREATE_CATALOG,
             eventMetadataFactory.create(),
             new EventAttributeMap()
                 .put(EventAttributes.CATALOG_NAME, request.getCatalog().getName())));
     Response resp = delegate.createCatalog(request, realmContext, securityContext);
-    polarisEventListener.onEvent(
+    polarisEventDispatcher.dispatch(
         new PolarisEvent(
             PolarisEventType.AFTER_CREATE_CATALOG,
             eventMetadataFactory.create(),
@@ -78,13 +78,13 @@ public Response createCatalog(
   @Override
   public Response deleteCatalog(
       String catalogName, RealmContext realmContext, SecurityContext securityContext) {
-    polarisEventListener.onEvent(
+    polarisEventDispatcher.dispatch(
         new PolarisEvent(
             PolarisEventType.BEFORE_DELETE_CATALOG,
             eventMetadataFactory.create(),
             new EventAttributeMap().put(EventAttributes.CATALOG_NAME, catalogName)));
     Response resp = delegate.deleteCatalog(catalogName, realmContext, securityContext);
-    polarisEventListener.onEvent(
+    polarisEventDispatcher.dispatch(
         new PolarisEvent(
             PolarisEventType.AFTER_DELETE_CATALOG,
             eventMetadataFactory.create(),
@@ -95,13 +95,13 @@ public Response deleteCatalog(
   @Override
   public Response getCatalog(
       String catalogName, RealmContext realmContext, SecurityContext securityContext) {
-    polarisEventListener.onEvent(
+    polarisEventDispatcher.dispatch(
         new PolarisEvent(
             PolarisEventType.BEFORE_GET_CATALOG,
             eventMetadataFactory.create(),
             new EventAttributeMap().put(EventAttributes.CATALOG_NAME, catalogName)));
     Response resp = delegate.getCatalog(catalogName, realmContext, securityContext);
-    polarisEventListener.onEvent(
+    polarisEventDispatcher.dispatch(
         new PolarisEvent(
             PolarisEventType.AFTER_GET_CATALOG,
             eventMetadataFactory.create(),
@@ -115,7 +115,7 @@ public Response updateCatalog(
       UpdateCatalogRequest updateRequest,
       RealmContext realmContext,
       SecurityContext securityContext) {
-    polarisEventListener.onEvent(
+    polarisEventDispatcher.dispatch(
         new PolarisEvent(
             PolarisEventType.BEFORE_UPDATE_CATALOG,
             eventMetadataFactory.create(),
@@ -124,7 +124,7 @@ public Response updateCatalog(
                 .put(EventAttributes.UPDATE_CATALOG_REQUEST, updateRequest)));
     Response resp =
         delegate.updateCatalog(catalogName, updateRequest, realmContext, securityContext);
-    polarisEventListener.onEvent(
+    polarisEventDispatcher.dispatch(
         new PolarisEvent(
             PolarisEventType.AFTER_UPDATE_CATALOG,
             eventMetadataFactory.create(),
@@ -134,13 +134,13 @@ public Response updateCatalog(
 
   @Override
   public Response listCatalogs(RealmContext realmContext, SecurityContext securityContext) {
-    polarisEventListener.onEvent(
+    polarisEventDispatcher.dispatch(
         new PolarisEvent(
             PolarisEventType.BEFORE_LIST_CATALOGS,
             eventMetadataFactory.create(),
             new EventAttributeMap()));
     Response resp = delegate.listCatalogs(realmContext, securityContext);
-    polarisEventListener.onEvent(
+    polarisEventDispatcher.dispatch(
         new PolarisEvent(
             PolarisEventType.AFTER_LIST_CATALOGS,
             eventMetadataFactory.create(),
@@ -154,15 +154,15 @@ public Response createCatalogRole(
       CreateCatalogRoleRequest request,
       RealmContext realmContext,
       SecurityContext securityContext) {
-    polarisEventListener.onEvent(
+    polarisEventDispatcher.dispatch(
         new PolarisEvent(
             PolarisEventType.BEFORE_CREATE_CATALOG_ROLE,
             eventMetadataFactory.create(),
             new EventAttributeMap()
                 .put(EventAttributes.CATALOG_NAME, catalogName)
                 .put(EventAttributes.CATALOG_ROLE_NAME, request.getCatalogRole().getName())));
     Response resp = delegate.createCatalogRole(catalogName, request, realmContext, securityContext);
-    polarisEventListener.onEvent(
+    polarisEventDispatcher.dispatch(
         new PolarisEvent(
             PolarisEventType.AFTER_CREATE_CATALOG_ROLE,
             eventMetadataFactory.create(),
@@ -178,7 +178,7 @@ public Response deleteCatalogRole(
       String catalogRoleName,
       RealmContext realmContext,
       SecurityContext securityContext) {
-    polarisEventListener.onEvent(
+    polarisEventDispatcher.dispatch(
         new PolarisEvent(
             PolarisEventType.BEFORE_DELETE_CATALOG_ROLE,
             eventMetadataFactory.create(),
@@ -187,7 +187,7 @@ public Response deleteCatalogRole(
                 .put(EventAttributes.CATALOG_ROLE_NAME, catalogRoleName)));
     Response resp =
         delegate.deleteCatalogRole(catalogName, catalogRoleName, realmContext, securityContext);
-    polarisEventListener.onEvent(
+    polarisEventDispatcher.dispatch(
         new PolarisEvent(
             PolarisEventType.AFTER_DELETE_CATALOG_ROLE,
             eventMetadataFactory.create(),
@@ -203,7 +203,7 @@ public Response getCatalogRole(
       String catalogRoleName,
       RealmContext realmContext,
       SecurityContext securityContext) {
-    polarisEventListener.onEvent(
+    polarisEventDispatcher.dispatch(
         new PolarisEvent(
             PolarisEventType.BEFORE_GET_CATALOG_ROLE,
             eventMetadataFactory.create(),
@@ -212,7 +212,7 @@ public Response getCatalogRole(
                 .put(EventAttributes.CATALOG_ROLE_NAME, catalogRoleName)));
     Response resp =
         delegate.getCatalogRole(catalogName, catalogRoleName, realmContext, securityContext);
-    polarisEventListener.onEvent(
+    polarisEventDispatcher.dispatch(
         new PolarisEvent(
             PolarisEventType.AFTER_GET_CATALOG_ROLE,
             eventMetadataFactory.create(),
@@ -229,7 +229,7 @@ public Response updateCatalogRole(
       UpdateCatalogRoleRequest updateRequest,
       RealmContext realmContext,
       SecurityContext securityContext) {
-    polarisEventListener.onEvent(
+    polarisEventDispatcher.dispatch(
         new PolarisEvent(
             PolarisEventType.BEFORE_UPDATE_CATALOG_ROLE,
             eventMetadataFactory.create(),
@@ -240,7 +240,7 @@ public Response updateCatalogRole(
     Response resp =
         delegate.updateCatalogRole(
             catalogName, catalogRoleName, updateRequest, realmContext, securityContext);
-    polarisEventListener.onEvent(
+    polarisEventDispatcher.dispatch(
         new PolarisEvent(
             PolarisEventType.AFTER_UPDATE_CATALOG_ROLE,
             eventMetadataFactory.create(),
@@ -253,13 +253,13 @@ public Response updateCatalogRole(
   @Override
   public Response listCatalogRoles(
       String catalogName, RealmContext realmContext, SecurityContext securityContext) {
-    polarisEventListener.onEvent(
+    polarisEventDispatcher.dispatch(
         new PolarisEvent(
             PolarisEventType.BEFORE_LIST_CATALOG_ROLES,
             eventMetadataFactory.create(),
             new EventAttributeMap().put(EventAttributes.CATALOG_NAME, catalogName)));
     Response resp = delegate.listCatalogRoles(catalogName, realmContext, securityContext);
-    polarisEventListener.onEvent(
+    polarisEventDispatcher.dispatch(
         new PolarisEvent(
             PolarisEventType.AFTER_LIST_CATALOG_ROLES,
             eventMetadataFactory.create(),
@@ -274,7 +274,7 @@ public Response addGrantToCatalogRole(
       AddGrantRequest grantRequest,
       RealmContext realmContext,
       SecurityContext securityContext) {
-    polarisEventListener.onEvent(
+    polarisEventDispatcher.dispatch(
         new PolarisEvent(
             PolarisEventType.BEFORE_ADD_GRANT_TO_CATALOG_ROLE,
             eventMetadataFactory.create(),
@@ -286,7 +286,7 @@ public Response addGrantToCatalogRole(
         delegate.addGrantToCatalogRole(
             catalogName, catalogRoleName, grantRequest, realmContext, securityContext);
     GrantResource grantResource = grantRequest.getGrant();
-    polarisEventListener.onEvent(
+    polarisEventDispatcher.dispatch(
         new PolarisEvent(
             PolarisEventType.AFTER_ADD_GRANT_TO_CATALOG_ROLE,
             eventMetadataFactory.create(),
@@ -306,7 +306,7 @@ public Response revokeGrantFromCatalogRole(
       RevokeGrantRequest grantRequest,
       RealmContext realmContext,
       SecurityContext securityContext) {
-    polarisEventListener.onEvent(
+    polarisEventDispatcher.dispatch(
         new PolarisEvent(
             PolarisEventType.BEFORE_REVOKE_GRANT_FROM_CATALOG_ROLE,
             eventMetadataFactory.create(),
@@ -319,7 +319,7 @@ public Response revokeGrantFromCatalogRole(
         delegate.revokeGrantFromCatalogRole(
             catalogName, catalogRoleName, cascade, grantRequest, realmContext, securityContext);
     GrantResource grantResource = grantRequest.getGrant();
-    polarisEventListener.onEvent(
+    polarisEventDispatcher.dispatch(
         new PolarisEvent(
             PolarisEventType.AFTER_REVOKE_GRANT_FROM_CATALOG_ROLE,
             eventMetadataFactory.create(),
@@ -338,7 +338,7 @@ public Response listAssigneePrincipalRolesForCatalogRole(
       String catalogRoleName,
       RealmContext realmContext,
       SecurityContext securityContext) {
-    polarisEventListener.onEvent(
+    polarisEventDispatcher.dispatch(
         new PolarisEvent(
             PolarisEventType.BEFORE_LIST_ASSIGNEE_PRINCIPAL_ROLES_FOR_CATALOG_ROLE,
             eventMetadataFactory.create(),
@@ -348,7 +348,7 @@ public Response listAssigneePrincipalRolesForCatalogRole(
     Response resp =
         delegate.listAssigneePrincipalRolesForCatalogRole(
             catalogName, catalogRoleName, realmContext, securityContext);
-    polarisEventListener.onEvent(
+    polarisEventDispatcher.dispatch(
         new PolarisEvent(
             PolarisEventType.AFTER_LIST_ASSIGNEE_PRINCIPAL_ROLES_FOR_CATALOG_ROLE,
             eventMetadataFactory.create(),
@@ -364,7 +364,7 @@ public Response listGrantsForCatalogRole(
       String catalogRoleName,
       RealmContext realmContext,
       SecurityContext securityContext) {
-    polarisEventListener.onEvent(
+    polarisEventDispatcher.dispatch(
         new PolarisEvent(
             PolarisEventType.BEFORE_LIST_GRANTS_FOR_CATALOG_ROLE,
             eventMetadataFactory.create(),
@@ -374,7 +374,7 @@ public Response listGrantsForCatalogRole(
     Response resp =
         delegate.listGrantsForCatalogRole(
             catalogName, catalogRoleName, realmContext, securityContext);
-    polarisEventListener.onEvent(
+    polarisEventDispatcher.dispatch(
         new PolarisEvent(
             PolarisEventType.AFTER_LIST_GRANTS_FOR_CATALOG_ROLE,
             eventMetadataFactory.create(),