You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@polaris.apache.org by yu...@apache.org on 2026/03/25 16:58:53 UTC
(polaris) branch main updated: Enable multiple event listeners to consume for Polaris events (#3973)
This is an automated email from the ASF dual-hosted git repository.
yufei pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/polaris.git
The following commit(s) were added to refs/heads/main by this push:
new e94fdff638 Enable multiple event listeners to consume for Polaris events (#3973)
e94fdff638 is described below
commit e94fdff63852dc41635c9e7eb62b3627ba562b85
Author: Nándor Kollár <na...@users.noreply.github.com>
AuthorDate: Wed Mar 25 17:58:45 2026 +0100
Enable multiple event listeners to consume for Polaris events (#3973)
---
CHANGELOG.md | 1 +
.../extension/auth/opa/test/OpaTestProfiles.java | 1 -
.../OAuthClientCredentialsParametersDpo.java | 4 -
.../src/main/resources/application.properties | 2 -
.../PolarisCatalogsEventServiceDelegator.java | 60 ++++++------
...PolarisPrincipalRolesEventServiceDelegator.java | 40 ++++----
.../PolarisPrincipalsEventServiceDelegator.java | 44 ++++-----
.../CatalogGenericTableEventServiceDelegator.java | 20 ++--
.../service/catalog/iceberg/IcebergCatalog.java | 16 +--
.../IcebergRestCatalogEventServiceDelegator.java | 108 ++++++++++-----------
...bergRestConfigurationEventServiceDelegator.java | 12 +--
.../policy/CatalogPolicyEventServiceDelegator.java | 36 +++----
.../polaris/service/config/ServiceProducers.java | 10 --
.../catalog/PolarisCallContextCatalogFactory.java | 10 +-
...sEventListener.java => EventBusConfigurer.java} | 18 +++-
...allbackEventListenerTypeConfigInterceptor.java} | 16 +--
...ntListener.java => PolarisEventDispatcher.java} | 11 +--
.../events/PolarisEventListenerConfiguration.java | 13 ++-
.../service/events/PolarisEventListeners.java | 53 ++++++++++
....java => PolarisServiceBusEventDispatcher.java} | 28 +++++-
.../service/ratelimiter/RateLimiterFilter.java | 10 +-
.../polaris/service/task/TaskExecutorImpl.java | 12 +--
.../io.smallrye.config.ConfigSourceInterceptor | 10 +-
.../service/admin/PolarisAuthzTestBase.java | 10 +-
.../AbstractPolarisGenericTableCatalogTest.java | 4 +-
.../AbstractIcebergCatalogHandlerAuthzTest.java | 2 +-
.../iceberg/AbstractIcebergCatalogTest.java | 12 ++-
.../iceberg/AbstractIcebergCatalogViewTest.java | 12 ++-
.../iceberg/CommitTransactionEventTest.java | 37 ++++---
.../service/catalog/io/FileIOFactoryTest.java | 2 +-
.../catalog/policy/AbstractPolicyCatalogTest.java | 4 +-
.../PolarisEventListenerOldConfigurationTest.java | 56 +++++++++++
...PolarisEventListenersMultipleListenersTest.java | 88 +++++++++++++++++
.../service/ratelimiter/RateLimiterFilterTest.java | 16 ++-
.../polaris/service/task/TaskExecutorImplTest.java | 12 +--
.../org/apache/polaris/service/TestServices.java | 16 +--
.../events/listeners/InMemoryEventCollector.java} | 32 ++++--
.../smallrye-polaris_event_listener.md | 3 +-
38 files changed, 550 insertions(+), 291 deletions(-)
diff --git a/CHANGELOG.md b/CHANGELOG.md
index ca4b4c325e..9a3457a8b8 100644
--- a/CHANGELOG.md
+++ b/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
diff --git a/extensions/auth/opa/tests/src/intTest/java/org/apache/polaris/extension/auth/opa/test/OpaTestProfiles.java b/extensions/auth/opa/tests/src/intTest/java/org/apache/polaris/extension/auth/opa/test/OpaTestProfiles.java
index 31276cce5a..a4864946af 100644
--- a/extensions/auth/opa/tests/src/intTest/java/org/apache/polaris/extension/auth/opa/test/OpaTestProfiles.java
+++ b/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;
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/connection/OAuthClientCredentialsParametersDpo.java b/polaris-core/src/main/java/org/apache/polaris/core/connection/OAuthClientCredentialsParametersDpo.java
index 7b855496b9..e6b9c9b19a 100644
--- a/polaris-core/src/main/java/org/apache/polaris/core/connection/OAuthClientCredentialsParametersDpo.java
+++ b/polaris-core/src/main/java/org/apache/polaris/core/connection/OAuthClientCredentialsParametersDpo.java
@@ -20,7 +20,6 @@ package org.apache.polaris.core.connection;
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 @@ import org.apache.polaris.core.secrets.SecretReference;
* API model.
*/
public class OAuthClientCredentialsParametersDpo extends AuthenticationParametersDpo {
-
- private static final Joiner COLON_JOINER = Joiner.on(":");
-
@JsonProperty(value = "tokenUri")
private final String tokenUri;
diff --git a/runtime/defaults/src/main/resources/application.properties b/runtime/defaults/src/main/resources/application.properties
index 8560a3cf0e..d0d4cc3916 100644
--- a/runtime/defaults/src/main/resources/application.properties
+++ b/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
diff --git a/runtime/service/src/main/java/org/apache/polaris/service/admin/PolarisCatalogsEventServiceDelegator.java b/runtime/service/src/main/java/org/apache/polaris/service/admin/PolarisCatalogsEventServiceDelegator.java
index c316a3d535..9ecdbdbe3c 100644
--- a/runtime/service/src/main/java/org/apache/polaris/service/admin/PolarisCatalogsEventServiceDelegator.java
+++ b/runtime/service/src/main/java/org/apache/polaris/service/admin/PolarisCatalogsEventServiceDelegator.java
@@ -45,29 +45,29 @@ import org.apache.polaris.service.admin.api.PolarisCatalogsApiService;
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 class PolarisCatalogsEventServiceDelegator implements PolarisCatalogsApiS
@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 class PolarisCatalogsEventServiceDelegator implements PolarisCatalogsApiS
@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 class PolarisCatalogsEventServiceDelegator implements PolarisCatalogsApiS
UpdateCatalogRequest updateRequest,
RealmContext realmContext,
SecurityContext securityContext) {
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_UPDATE_CATALOG,
eventMetadataFactory.create(),
@@ -124,7 +124,7 @@ public class PolarisCatalogsEventServiceDelegator implements PolarisCatalogsApiS
.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 class PolarisCatalogsEventServiceDelegator implements PolarisCatalogsApiS
@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,7 +154,7 @@ public class PolarisCatalogsEventServiceDelegator implements PolarisCatalogsApiS
CreateCatalogRoleRequest request,
RealmContext realmContext,
SecurityContext securityContext) {
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_CREATE_CATALOG_ROLE,
eventMetadataFactory.create(),
@@ -162,7 +162,7 @@ public class PolarisCatalogsEventServiceDelegator implements PolarisCatalogsApiS
.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 class PolarisCatalogsEventServiceDelegator implements PolarisCatalogsApiS
String catalogRoleName,
RealmContext realmContext,
SecurityContext securityContext) {
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_DELETE_CATALOG_ROLE,
eventMetadataFactory.create(),
@@ -187,7 +187,7 @@ public class PolarisCatalogsEventServiceDelegator implements PolarisCatalogsApiS
.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 class PolarisCatalogsEventServiceDelegator implements PolarisCatalogsApiS
String catalogRoleName,
RealmContext realmContext,
SecurityContext securityContext) {
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_GET_CATALOG_ROLE,
eventMetadataFactory.create(),
@@ -212,7 +212,7 @@ public class PolarisCatalogsEventServiceDelegator implements PolarisCatalogsApiS
.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 class PolarisCatalogsEventServiceDelegator implements PolarisCatalogsApiS
UpdateCatalogRoleRequest updateRequest,
RealmContext realmContext,
SecurityContext securityContext) {
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_UPDATE_CATALOG_ROLE,
eventMetadataFactory.create(),
@@ -240,7 +240,7 @@ public class PolarisCatalogsEventServiceDelegator implements PolarisCatalogsApiS
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 class PolarisCatalogsEventServiceDelegator implements PolarisCatalogsApiS
@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 class PolarisCatalogsEventServiceDelegator implements PolarisCatalogsApiS
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 class PolarisCatalogsEventServiceDelegator implements PolarisCatalogsApiS
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 class PolarisCatalogsEventServiceDelegator implements PolarisCatalogsApiS
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 class PolarisCatalogsEventServiceDelegator implements PolarisCatalogsApiS
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 class PolarisCatalogsEventServiceDelegator implements PolarisCatalogsApiS
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 class PolarisCatalogsEventServiceDelegator implements PolarisCatalogsApiS
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 class PolarisCatalogsEventServiceDelegator implements PolarisCatalogsApiS
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 class PolarisCatalogsEventServiceDelegator implements PolarisCatalogsApiS
Response resp =
delegate.listGrantsForCatalogRole(
catalogName, catalogRoleName, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_LIST_GRANTS_FOR_CATALOG_ROLE,
eventMetadataFactory.create(),
diff --git a/runtime/service/src/main/java/org/apache/polaris/service/admin/PolarisPrincipalRolesEventServiceDelegator.java b/runtime/service/src/main/java/org/apache/polaris/service/admin/PolarisPrincipalRolesEventServiceDelegator.java
index b26c52913d..4544159b64 100644
--- a/runtime/service/src/main/java/org/apache/polaris/service/admin/PolarisPrincipalRolesEventServiceDelegator.java
+++ b/runtime/service/src/main/java/org/apache/polaris/service/admin/PolarisPrincipalRolesEventServiceDelegator.java
@@ -34,16 +34,16 @@ import org.apache.polaris.service.admin.api.PolarisPrincipalRolesApiService;
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 PolarisPrincipalRolesEventServiceDelegator implements PolarisPrincipalRolesApiService {
@Inject @Delegate PolarisPrincipalRolesApiService delegate;
- @Inject PolarisEventListener polarisEventListener;
+ @Inject PolarisEventDispatcher polarisEventDispatcher;
@Inject PolarisEventMetadataFactory eventMetadataFactory;
@Override
@@ -51,13 +51,13 @@ public class PolarisPrincipalRolesEventServiceDelegator implements PolarisPrinci
CreatePrincipalRoleRequest request,
RealmContext realmContext,
SecurityContext securityContext) {
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_CREATE_PRINCIPAL_ROLE,
eventMetadataFactory.create(),
new EventAttributeMap().put(EventAttributes.CREATE_PRINCIPAL_ROLE_REQUEST, request)));
Response resp = delegate.createPrincipalRole(request, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_CREATE_PRINCIPAL_ROLE,
eventMetadataFactory.create(),
@@ -69,13 +69,13 @@ public class PolarisPrincipalRolesEventServiceDelegator implements PolarisPrinci
@Override
public Response deletePrincipalRole(
String principalRoleName, RealmContext realmContext, SecurityContext securityContext) {
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_DELETE_PRINCIPAL_ROLE,
eventMetadataFactory.create(),
new EventAttributeMap().put(EventAttributes.PRINCIPAL_ROLE_NAME, principalRoleName)));
Response resp = delegate.deletePrincipalRole(principalRoleName, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_DELETE_PRINCIPAL_ROLE,
eventMetadataFactory.create(),
@@ -86,13 +86,13 @@ public class PolarisPrincipalRolesEventServiceDelegator implements PolarisPrinci
@Override
public Response getPrincipalRole(
String principalRoleName, RealmContext realmContext, SecurityContext securityContext) {
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_GET_PRINCIPAL_ROLE,
eventMetadataFactory.create(),
new EventAttributeMap().put(EventAttributes.PRINCIPAL_ROLE_NAME, principalRoleName)));
Response resp = delegate.getPrincipalRole(principalRoleName, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_GET_PRINCIPAL_ROLE,
eventMetadataFactory.create(),
@@ -107,7 +107,7 @@ public class PolarisPrincipalRolesEventServiceDelegator implements PolarisPrinci
UpdatePrincipalRoleRequest updateRequest,
RealmContext realmContext,
SecurityContext securityContext) {
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_UPDATE_PRINCIPAL_ROLE,
eventMetadataFactory.create(),
@@ -117,7 +117,7 @@ public class PolarisPrincipalRolesEventServiceDelegator implements PolarisPrinci
Response resp =
delegate.updatePrincipalRole(
principalRoleName, updateRequest, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_UPDATE_PRINCIPAL_ROLE,
eventMetadataFactory.create(),
@@ -128,13 +128,13 @@ public class PolarisPrincipalRolesEventServiceDelegator implements PolarisPrinci
@Override
public Response listPrincipalRoles(RealmContext realmContext, SecurityContext securityContext) {
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_LIST_PRINCIPAL_ROLES,
eventMetadataFactory.create(),
new EventAttributeMap()));
Response resp = delegate.listPrincipalRoles(realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_LIST_PRINCIPAL_ROLES,
eventMetadataFactory.create(),
@@ -149,7 +149,7 @@ public class PolarisPrincipalRolesEventServiceDelegator implements PolarisPrinci
GrantCatalogRoleRequest request,
RealmContext realmContext,
SecurityContext securityContext) {
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_ASSIGN_CATALOG_ROLE_TO_PRINCIPAL_ROLE,
eventMetadataFactory.create(),
@@ -160,7 +160,7 @@ public class PolarisPrincipalRolesEventServiceDelegator implements PolarisPrinci
Response resp =
delegate.assignCatalogRoleToPrincipalRole(
principalRoleName, catalogName, request, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_ASSIGN_CATALOG_ROLE_TO_PRINCIPAL_ROLE,
eventMetadataFactory.create(),
@@ -178,7 +178,7 @@ public class PolarisPrincipalRolesEventServiceDelegator implements PolarisPrinci
String catalogRoleName,
RealmContext realmContext,
SecurityContext securityContext) {
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_REVOKE_CATALOG_ROLE_FROM_PRINCIPAL_ROLE,
eventMetadataFactory.create(),
@@ -189,7 +189,7 @@ public class PolarisPrincipalRolesEventServiceDelegator implements PolarisPrinci
Response resp =
delegate.revokeCatalogRoleFromPrincipalRole(
principalRoleName, catalogName, catalogRoleName, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_REVOKE_CATALOG_ROLE_FROM_PRINCIPAL_ROLE,
eventMetadataFactory.create(),
@@ -203,7 +203,7 @@ public class PolarisPrincipalRolesEventServiceDelegator implements PolarisPrinci
@Override
public Response listAssigneePrincipalsForPrincipalRole(
String principalRoleName, RealmContext realmContext, SecurityContext securityContext) {
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_LIST_ASSIGNEE_PRINCIPALS_FOR_PRINCIPAL_ROLE,
eventMetadataFactory.create(),
@@ -211,7 +211,7 @@ public class PolarisPrincipalRolesEventServiceDelegator implements PolarisPrinci
Response resp =
delegate.listAssigneePrincipalsForPrincipalRole(
principalRoleName, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_LIST_ASSIGNEE_PRINCIPALS_FOR_PRINCIPAL_ROLE,
eventMetadataFactory.create(),
@@ -225,7 +225,7 @@ public class PolarisPrincipalRolesEventServiceDelegator implements PolarisPrinci
String catalogName,
RealmContext realmContext,
SecurityContext securityContext) {
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_LIST_CATALOG_ROLES_FOR_PRINCIPAL_ROLE,
eventMetadataFactory.create(),
@@ -235,7 +235,7 @@ public class PolarisPrincipalRolesEventServiceDelegator implements PolarisPrinci
Response resp =
delegate.listCatalogRolesForPrincipalRole(
principalRoleName, catalogName, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_LIST_CATALOG_ROLES_FOR_PRINCIPAL_ROLE,
eventMetadataFactory.create(),
diff --git a/runtime/service/src/main/java/org/apache/polaris/service/admin/PolarisPrincipalsEventServiceDelegator.java b/runtime/service/src/main/java/org/apache/polaris/service/admin/PolarisPrincipalsEventServiceDelegator.java
index fe156d758e..8c9f92d6d4 100644
--- a/runtime/service/src/main/java/org/apache/polaris/service/admin/PolarisPrincipalsEventServiceDelegator.java
+++ b/runtime/service/src/main/java/org/apache/polaris/service/admin/PolarisPrincipalsEventServiceDelegator.java
@@ -36,29 +36,29 @@ import org.apache.polaris.service.admin.api.PolarisPrincipalsApiService;
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 PolarisPrincipalsEventServiceDelegator implements PolarisPrincipalsApiService {
@Inject @Delegate PolarisPrincipalsApiService delegate;
- @Inject PolarisEventListener polarisEventListener;
+ @Inject PolarisEventDispatcher polarisEventDispatcher;
@Inject PolarisEventMetadataFactory eventMetadataFactory;
@Override
public Response createPrincipal(
CreatePrincipalRequest request, RealmContext realmContext, SecurityContext securityContext) {
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_CREATE_PRINCIPAL,
eventMetadataFactory.create(),
new EventAttributeMap()
.put(EventAttributes.PRINCIPAL_NAME, request.getPrincipal().getName())));
Response resp = delegate.createPrincipal(request, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_CREATE_PRINCIPAL,
eventMetadataFactory.create(),
@@ -75,7 +75,7 @@ public class PolarisPrincipalsEventServiceDelegator implements PolarisPrincipals
ResetPrincipalRequest resetPrincipalRequest,
RealmContext realmContext,
SecurityContext securityContext) {
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_RESET_CREDENTIALS,
eventMetadataFactory.create(),
@@ -83,7 +83,7 @@ public class PolarisPrincipalsEventServiceDelegator implements PolarisPrincipals
Response resp =
delegate.resetCredentials(
principalName, resetPrincipalRequest, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_RESET_CREDENTIALS,
eventMetadataFactory.create(),
@@ -97,13 +97,13 @@ public class PolarisPrincipalsEventServiceDelegator implements PolarisPrincipals
@Override
public Response deletePrincipal(
String principalName, RealmContext realmContext, SecurityContext securityContext) {
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_DELETE_PRINCIPAL,
eventMetadataFactory.create(),
new EventAttributeMap().put(EventAttributes.PRINCIPAL_NAME, principalName)));
Response resp = delegate.deletePrincipal(principalName, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_DELETE_PRINCIPAL,
eventMetadataFactory.create(),
@@ -114,13 +114,13 @@ public class PolarisPrincipalsEventServiceDelegator implements PolarisPrincipals
@Override
public Response getPrincipal(
String principalName, RealmContext realmContext, SecurityContext securityContext) {
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_GET_PRINCIPAL,
eventMetadataFactory.create(),
new EventAttributeMap().put(EventAttributes.PRINCIPAL_NAME, principalName)));
Response resp = delegate.getPrincipal(principalName, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_GET_PRINCIPAL,
eventMetadataFactory.create(),
@@ -134,7 +134,7 @@ public class PolarisPrincipalsEventServiceDelegator implements PolarisPrincipals
UpdatePrincipalRequest updateRequest,
RealmContext realmContext,
SecurityContext securityContext) {
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_UPDATE_PRINCIPAL,
eventMetadataFactory.create(),
@@ -143,7 +143,7 @@ public class PolarisPrincipalsEventServiceDelegator implements PolarisPrincipals
.put(EventAttributes.UPDATE_PRINCIPAL_REQUEST, updateRequest)));
Response resp =
delegate.updatePrincipal(principalName, updateRequest, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_UPDATE_PRINCIPAL,
eventMetadataFactory.create(),
@@ -154,14 +154,14 @@ public class PolarisPrincipalsEventServiceDelegator implements PolarisPrincipals
@Override
public Response rotateCredentials(
String principalName, RealmContext realmContext, SecurityContext securityContext) {
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_ROTATE_CREDENTIALS,
eventMetadataFactory.create(),
new EventAttributeMap().put(EventAttributes.PRINCIPAL_NAME, principalName)));
Response resp = delegate.rotateCredentials(principalName, realmContext, securityContext);
PrincipalWithCredentials principalWithCredentials = (PrincipalWithCredentials) resp.getEntity();
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_ROTATE_CREDENTIALS,
eventMetadataFactory.create(),
@@ -172,13 +172,13 @@ public class PolarisPrincipalsEventServiceDelegator implements PolarisPrincipals
@Override
public Response listPrincipals(RealmContext realmContext, SecurityContext securityContext) {
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_LIST_PRINCIPALS,
eventMetadataFactory.create(),
new EventAttributeMap()));
Response resp = delegate.listPrincipals(realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_LIST_PRINCIPALS,
eventMetadataFactory.create(),
@@ -192,7 +192,7 @@ public class PolarisPrincipalsEventServiceDelegator implements PolarisPrincipals
GrantPrincipalRoleRequest request,
RealmContext realmContext,
SecurityContext securityContext) {
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_ASSIGN_PRINCIPAL_ROLE,
eventMetadataFactory.create(),
@@ -201,7 +201,7 @@ public class PolarisPrincipalsEventServiceDelegator implements PolarisPrincipals
.put(EventAttributes.PRINCIPAL_ROLE, request.getPrincipalRole())));
Response resp =
delegate.assignPrincipalRole(principalName, request, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_ASSIGN_PRINCIPAL_ROLE,
eventMetadataFactory.create(),
@@ -217,7 +217,7 @@ public class PolarisPrincipalsEventServiceDelegator implements PolarisPrincipals
String principalRoleName,
RealmContext realmContext,
SecurityContext securityContext) {
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_REVOKE_PRINCIPAL_ROLE,
eventMetadataFactory.create(),
@@ -227,7 +227,7 @@ public class PolarisPrincipalsEventServiceDelegator implements PolarisPrincipals
Response resp =
delegate.revokePrincipalRole(
principalName, principalRoleName, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_REVOKE_PRINCIPAL_ROLE,
eventMetadataFactory.create(),
@@ -240,14 +240,14 @@ public class PolarisPrincipalsEventServiceDelegator implements PolarisPrincipals
@Override
public Response listPrincipalRolesAssigned(
String principalName, RealmContext realmContext, SecurityContext securityContext) {
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_LIST_ASSIGNED_PRINCIPAL_ROLES,
eventMetadataFactory.create(),
new EventAttributeMap().put(EventAttributes.PRINCIPAL_NAME, principalName)));
Response resp =
delegate.listPrincipalRolesAssigned(principalName, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_LIST_ASSIGNED_PRINCIPAL_ROLES,
eventMetadataFactory.create(),
diff --git a/runtime/service/src/main/java/org/apache/polaris/service/catalog/generic/CatalogGenericTableEventServiceDelegator.java b/runtime/service/src/main/java/org/apache/polaris/service/catalog/generic/CatalogGenericTableEventServiceDelegator.java
index 9b116b9b34..2cecbae0bd 100644
--- a/runtime/service/src/main/java/org/apache/polaris/service/catalog/generic/CatalogGenericTableEventServiceDelegator.java
+++ b/runtime/service/src/main/java/org/apache/polaris/service/catalog/generic/CatalogGenericTableEventServiceDelegator.java
@@ -32,9 +32,9 @@ import org.apache.polaris.service.catalog.common.CatalogAdapter;
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;
import org.apache.polaris.service.types.CreateGenericTableRequest;
import org.apache.polaris.service.types.LoadGenericTableResponse;
@@ -44,7 +44,7 @@ public class CatalogGenericTableEventServiceDelegator
implements PolarisCatalogGenericTableApiService, CatalogAdapter {
@Inject @Delegate GenericTableCatalogAdapter delegate;
- @Inject PolarisEventListener polarisEventListener;
+ @Inject PolarisEventDispatcher polarisEventDispatcher;
@Inject PolarisEventMetadataFactory eventMetadataFactory;
@Inject CatalogPrefixParser prefixParser;
@@ -56,7 +56,7 @@ public class CatalogGenericTableEventServiceDelegator
RealmContext realmContext,
SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_CREATE_GENERIC_TABLE,
eventMetadataFactory.create(),
@@ -67,7 +67,7 @@ public class CatalogGenericTableEventServiceDelegator
Response resp =
delegate.createGenericTable(
prefix, namespace, createGenericTableRequest, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_CREATE_GENERIC_TABLE,
eventMetadataFactory.create(),
@@ -88,7 +88,7 @@ public class CatalogGenericTableEventServiceDelegator
RealmContext realmContext,
SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_DROP_GENERIC_TABLE,
eventMetadataFactory.create(),
@@ -98,7 +98,7 @@ public class CatalogGenericTableEventServiceDelegator
.put(EventAttributes.GENERIC_TABLE_NAME, genericTable)));
Response resp =
delegate.dropGenericTable(prefix, namespace, genericTable, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_DROP_GENERIC_TABLE,
eventMetadataFactory.create(),
@@ -118,7 +118,7 @@ public class CatalogGenericTableEventServiceDelegator
RealmContext realmContext,
SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_LIST_GENERIC_TABLES,
eventMetadataFactory.create(),
@@ -128,7 +128,7 @@ public class CatalogGenericTableEventServiceDelegator
Response resp =
delegate.listGenericTables(
prefix, namespace, pageToken, pageSize, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_LIST_GENERIC_TABLES,
eventMetadataFactory.create(),
@@ -146,7 +146,7 @@ public class CatalogGenericTableEventServiceDelegator
RealmContext realmContext,
SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_LOAD_GENERIC_TABLE,
eventMetadataFactory.create(),
@@ -156,7 +156,7 @@ public class CatalogGenericTableEventServiceDelegator
.put(EventAttributes.GENERIC_TABLE_NAME, genericTable)));
Response resp =
delegate.loadGenericTable(prefix, namespace, genericTable, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_LOAD_GENERIC_TABLE,
eventMetadataFactory.create(),
diff --git a/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalog.java b/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalog.java
index 5cd5490000..28b6ab12b1 100644
--- a/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalog.java
+++ b/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalog.java
@@ -141,9 +141,9 @@ import org.apache.polaris.service.catalog.validation.IcebergPropertiesValidation
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;
import org.apache.polaris.service.task.TaskExecutor;
import org.apache.polaris.service.types.NotificationRequest;
import org.apache.polaris.service.types.NotificationType;
@@ -179,7 +179,7 @@ public class IcebergCatalog extends BaseMetastoreViewCatalog
private final CatalogEntity catalogEntity;
private final TaskExecutor taskExecutor;
private final PolarisPrincipal principal;
- private final PolarisEventListener polarisEventListener;
+ private final PolarisEventDispatcher polarisEventDispatcher;
private final PolarisEventMetadataFactory eventMetadataFactory;
private final AtomicBoolean loggedPrefixOverlapWarning = new AtomicBoolean(false);
@@ -212,7 +212,7 @@ public class IcebergCatalog extends BaseMetastoreViewCatalog
TaskExecutor taskExecutor,
StorageAccessConfigProvider storageAccessConfigProvider,
FileIOFactory fileIOFactory,
- PolarisEventListener polarisEventListener,
+ PolarisEventDispatcher polarisEventDispatcher,
PolarisEventMetadataFactory eventMetadataFactory) {
this.diagnostics = diagnostics;
this.resolverFactory = resolverFactory;
@@ -227,7 +227,7 @@ public class IcebergCatalog extends BaseMetastoreViewCatalog
this.storageAccessConfigProvider = storageAccessConfigProvider;
this.fileIOFactory = fileIOFactory;
this.metaStoreManager = metaStoreManager;
- this.polarisEventListener = polarisEventListener;
+ this.polarisEventDispatcher = polarisEventDispatcher;
this.eventMetadataFactory = eventMetadataFactory;
}
@@ -1468,7 +1468,7 @@ public class IcebergCatalog extends BaseMetastoreViewCatalog
if (latestLocation == null) {
disableRefresh();
} else {
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_REFRESH_TABLE,
eventMetadataFactory.create(),
@@ -1494,7 +1494,7 @@ public class IcebergCatalog extends BaseMetastoreViewCatalog
Set.of(PolarisStorageActions.READ, PolarisStorageActions.LIST));
return TableMetadataParser.read(fileIO, metadataLocation);
});
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_REFRESH_TABLE,
eventMetadataFactory.create(),
@@ -1891,7 +1891,7 @@ public class IcebergCatalog extends BaseMetastoreViewCatalog
if (latestLocation == null) {
disableRefresh();
} else {
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_REFRESH_VIEW,
eventMetadataFactory.create(),
@@ -1919,7 +1919,7 @@ public class IcebergCatalog extends BaseMetastoreViewCatalog
return ViewMetadataParser.read(fileIO.newInputFile(metadataLocation));
});
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_REFRESH_VIEW,
eventMetadataFactory.create(),
diff --git a/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergRestCatalogEventServiceDelegator.java b/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergRestCatalogEventServiceDelegator.java
index 7f3f672336..f8a09b9033 100644
--- a/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergRestCatalogEventServiceDelegator.java
+++ b/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergRestCatalogEventServiceDelegator.java
@@ -52,9 +52,9 @@ import org.apache.polaris.service.catalog.common.CatalogAdapter;
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;
import org.apache.polaris.service.types.CommitTableRequest;
import org.apache.polaris.service.types.CommitViewRequest;
import org.apache.polaris.service.types.NotificationRequest;
@@ -65,7 +65,7 @@ public class IcebergRestCatalogEventServiceDelegator
implements IcebergRestCatalogApiService, CatalogAdapter {
@Inject @Delegate IcebergCatalogAdapter delegate;
- @Inject PolarisEventListener polarisEventListener;
+ @Inject PolarisEventDispatcher polarisEventDispatcher;
@Inject PolarisEventMetadataFactory eventMetadataFactory;
@Inject CatalogPrefixParser prefixParser;
@Inject EventAttributeMap eventAttributeMap;
@@ -74,12 +74,12 @@ public class IcebergRestCatalogEventServiceDelegator
@VisibleForTesting
public IcebergRestCatalogEventServiceDelegator(
IcebergCatalogAdapter delegate,
- PolarisEventListener polarisEventListener,
+ PolarisEventDispatcher polarisEventDispatcher,
PolarisEventMetadataFactory eventMetadataFactory,
CatalogPrefixParser prefixParser,
EventAttributeMap eventAttributeMap) {
this.delegate = delegate;
- this.polarisEventListener = polarisEventListener;
+ this.polarisEventDispatcher = polarisEventDispatcher;
this.eventMetadataFactory = eventMetadataFactory;
this.prefixParser = prefixParser;
this.eventAttributeMap = eventAttributeMap;
@@ -95,7 +95,7 @@ public class IcebergRestCatalogEventServiceDelegator
RealmContext realmContext,
SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_CREATE_NAMESPACE,
eventMetadataFactory.create(),
@@ -105,7 +105,7 @@ public class IcebergRestCatalogEventServiceDelegator
Response resp =
delegate.createNamespace(prefix, createNamespaceRequest, realmContext, securityContext);
CreateNamespaceResponse createNamespaceResponse = (CreateNamespaceResponse) resp.getEntity();
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_CREATE_NAMESPACE,
eventMetadataFactory.create(),
@@ -125,7 +125,7 @@ public class IcebergRestCatalogEventServiceDelegator
RealmContext realmContext,
SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_LIST_NAMESPACES,
eventMetadataFactory.create(),
@@ -134,7 +134,7 @@ public class IcebergRestCatalogEventServiceDelegator
.put(EventAttributes.PARENT_NAMESPACE_FQN, parent)));
Response resp =
delegate.listNamespaces(prefix, pageToken, pageSize, parent, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_LIST_NAMESPACES,
eventMetadataFactory.create(),
@@ -148,7 +148,7 @@ public class IcebergRestCatalogEventServiceDelegator
public Response loadNamespaceMetadata(
String prefix, String namespace, RealmContext realmContext, SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_LOAD_NAMESPACE_METADATA,
eventMetadataFactory.create(),
@@ -158,7 +158,7 @@ public class IcebergRestCatalogEventServiceDelegator
Response resp =
delegate.loadNamespaceMetadata(prefix, namespace, realmContext, securityContext);
GetNamespaceResponse getNamespaceResponse = (GetNamespaceResponse) resp.getEntity();
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_LOAD_NAMESPACE_METADATA,
eventMetadataFactory.create(),
@@ -174,7 +174,7 @@ public class IcebergRestCatalogEventServiceDelegator
String prefix, String namespace, RealmContext realmContext, SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
Namespace namespaceObj = decodeNamespace(namespace);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_CHECK_EXISTS_NAMESPACE,
eventMetadataFactory.create(),
@@ -182,7 +182,7 @@ public class IcebergRestCatalogEventServiceDelegator
.put(EventAttributes.CATALOG_NAME, catalogName)
.put(EventAttributes.NAMESPACE, namespaceObj)));
Response resp = delegate.namespaceExists(prefix, namespace, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_CHECK_EXISTS_NAMESPACE,
eventMetadataFactory.create(),
@@ -196,7 +196,7 @@ public class IcebergRestCatalogEventServiceDelegator
public Response dropNamespace(
String prefix, String namespace, RealmContext realmContext, SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_DROP_NAMESPACE,
eventMetadataFactory.create(),
@@ -204,7 +204,7 @@ public class IcebergRestCatalogEventServiceDelegator
.put(EventAttributes.CATALOG_NAME, catalogName)
.put(EventAttributes.NAMESPACE, decodeNamespace(namespace))));
Response resp = delegate.dropNamespace(prefix, namespace, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_DROP_NAMESPACE,
eventMetadataFactory.create(),
@@ -223,7 +223,7 @@ public class IcebergRestCatalogEventServiceDelegator
SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
Namespace namespaceObj = decodeNamespace(namespace);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_UPDATE_NAMESPACE_PROPERTIES,
eventMetadataFactory.create(),
@@ -236,7 +236,7 @@ public class IcebergRestCatalogEventServiceDelegator
Response resp =
delegate.updateProperties(
prefix, namespace, updateNamespacePropertiesRequest, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_UPDATE_NAMESPACE_PROPERTIES,
eventMetadataFactory.create(),
@@ -259,7 +259,7 @@ public class IcebergRestCatalogEventServiceDelegator
SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
Namespace namespaceObj = decodeNamespace(namespace);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_CREATE_TABLE,
eventMetadataFactory.create(),
@@ -277,7 +277,7 @@ public class IcebergRestCatalogEventServiceDelegator
realmContext,
securityContext);
if (!createTableRequest.stageCreate()) {
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_CREATE_TABLE,
eventMetadataFactory.create(),
@@ -300,7 +300,7 @@ public class IcebergRestCatalogEventServiceDelegator
SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
Namespace namespaceObj = decodeNamespace(namespace);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_LIST_TABLES,
eventMetadataFactory.create(),
@@ -309,7 +309,7 @@ public class IcebergRestCatalogEventServiceDelegator
.put(EventAttributes.NAMESPACE, namespaceObj)));
Response resp =
delegate.listTables(prefix, namespace, pageToken, pageSize, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_LIST_TABLES,
eventMetadataFactory.create(),
@@ -331,7 +331,7 @@ public class IcebergRestCatalogEventServiceDelegator
SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
Namespace namespaceObj = decodeNamespace(namespace);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_LOAD_TABLE,
eventMetadataFactory.create(),
@@ -353,7 +353,7 @@ public class IcebergRestCatalogEventServiceDelegator
realmContext,
securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_LOAD_TABLE,
eventMetadataFactory.create(),
@@ -374,7 +374,7 @@ public class IcebergRestCatalogEventServiceDelegator
SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
Namespace namespaceObj = decodeNamespace(namespace);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_CHECK_EXISTS_TABLE,
eventMetadataFactory.create(),
@@ -383,7 +383,7 @@ public class IcebergRestCatalogEventServiceDelegator
.put(EventAttributes.NAMESPACE, namespaceObj)
.put(EventAttributes.TABLE_NAME, table)));
Response resp = delegate.tableExists(prefix, namespace, table, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_CHECK_EXISTS_TABLE,
eventMetadataFactory.create(),
@@ -404,7 +404,7 @@ public class IcebergRestCatalogEventServiceDelegator
SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
Namespace namespaceObj = decodeNamespace(namespace);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_DROP_TABLE,
eventMetadataFactory.create(),
@@ -415,7 +415,7 @@ public class IcebergRestCatalogEventServiceDelegator
.put(EventAttributes.PURGE_REQUESTED, purgeRequested)));
Response resp =
delegate.dropTable(prefix, namespace, table, purgeRequested, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_DROP_TABLE,
eventMetadataFactory.create(),
@@ -436,7 +436,7 @@ public class IcebergRestCatalogEventServiceDelegator
SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
Namespace namespaceObj = decodeNamespace(namespace);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_REGISTER_TABLE,
eventMetadataFactory.create(),
@@ -447,7 +447,7 @@ public class IcebergRestCatalogEventServiceDelegator
Response resp =
delegate.registerTable(
prefix, namespace, registerTableRequest, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_REGISTER_TABLE,
eventMetadataFactory.create(),
@@ -466,7 +466,7 @@ public class IcebergRestCatalogEventServiceDelegator
RealmContext realmContext,
SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_RENAME_TABLE,
eventMetadataFactory.create(),
@@ -474,7 +474,7 @@ public class IcebergRestCatalogEventServiceDelegator
.put(EventAttributes.CATALOG_NAME, catalogName)
.put(EventAttributes.RENAME_TABLE_REQUEST, renameTableRequest)));
Response resp = delegate.renameTable(prefix, renameTableRequest, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_RENAME_TABLE,
eventMetadataFactory.create(),
@@ -494,7 +494,7 @@ public class IcebergRestCatalogEventServiceDelegator
SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
Namespace namespaceObj = decodeNamespace(namespace);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_UPDATE_TABLE,
eventMetadataFactory.create(),
@@ -506,7 +506,7 @@ public class IcebergRestCatalogEventServiceDelegator
Response resp =
delegate.updateTable(
prefix, namespace, table, commitTableRequest, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_UPDATE_TABLE,
eventMetadataFactory.create(),
@@ -530,7 +530,7 @@ public class IcebergRestCatalogEventServiceDelegator
SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
Namespace namespaceObj = decodeNamespace(namespace);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_CREATE_VIEW,
eventMetadataFactory.create(),
@@ -540,7 +540,7 @@ public class IcebergRestCatalogEventServiceDelegator
.put(EventAttributes.CREATE_VIEW_REQUEST, createViewRequest)));
Response resp =
delegate.createView(prefix, namespace, createViewRequest, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_CREATE_VIEW,
eventMetadataFactory.create(),
@@ -562,7 +562,7 @@ public class IcebergRestCatalogEventServiceDelegator
SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
Namespace namespaceObj = decodeNamespace(namespace);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_LIST_VIEWS,
eventMetadataFactory.create(),
@@ -571,7 +571,7 @@ public class IcebergRestCatalogEventServiceDelegator
.put(EventAttributes.NAMESPACE, namespaceObj)));
Response resp =
delegate.listViews(prefix, namespace, pageToken, pageSize, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_LIST_VIEWS,
eventMetadataFactory.create(),
@@ -590,7 +590,7 @@ public class IcebergRestCatalogEventServiceDelegator
SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
Namespace namespaceObj = decodeNamespace(namespace);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_LOAD_CREDENTIALS,
eventMetadataFactory.create(),
@@ -600,7 +600,7 @@ public class IcebergRestCatalogEventServiceDelegator
.put(EventAttributes.TABLE_NAME, table)));
Response resp =
delegate.loadCredentials(prefix, namespace, table, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_LOAD_CREDENTIALS,
eventMetadataFactory.create(),
@@ -620,7 +620,7 @@ public class IcebergRestCatalogEventServiceDelegator
SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
Namespace namespaceObj = decodeNamespace(namespace);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_LOAD_VIEW,
eventMetadataFactory.create(),
@@ -629,7 +629,7 @@ public class IcebergRestCatalogEventServiceDelegator
.put(EventAttributes.NAMESPACE, namespaceObj)
.put(EventAttributes.VIEW_NAME, view)));
Response resp = delegate.loadView(prefix, namespace, view, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_LOAD_VIEW,
eventMetadataFactory.create(),
@@ -650,7 +650,7 @@ public class IcebergRestCatalogEventServiceDelegator
SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
Namespace namespaceObj = decodeNamespace(namespace);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_CHECK_EXISTS_VIEW,
eventMetadataFactory.create(),
@@ -659,7 +659,7 @@ public class IcebergRestCatalogEventServiceDelegator
.put(EventAttributes.NAMESPACE, namespaceObj)
.put(EventAttributes.VIEW_NAME, view)));
Response resp = delegate.viewExists(prefix, namespace, view, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_CHECK_EXISTS_VIEW,
eventMetadataFactory.create(),
@@ -679,7 +679,7 @@ public class IcebergRestCatalogEventServiceDelegator
SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
Namespace namespaceObj = decodeNamespace(namespace);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_DROP_VIEW,
eventMetadataFactory.create(),
@@ -688,7 +688,7 @@ public class IcebergRestCatalogEventServiceDelegator
.put(EventAttributes.NAMESPACE, namespaceObj)
.put(EventAttributes.VIEW_NAME, view)));
Response resp = delegate.dropView(prefix, namespace, view, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_DROP_VIEW,
eventMetadataFactory.create(),
@@ -706,7 +706,7 @@ public class IcebergRestCatalogEventServiceDelegator
RealmContext realmContext,
SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_RENAME_VIEW,
eventMetadataFactory.create(),
@@ -714,7 +714,7 @@ public class IcebergRestCatalogEventServiceDelegator
.put(EventAttributes.CATALOG_NAME, catalogName)
.put(EventAttributes.RENAME_TABLE_REQUEST, renameTableRequest)));
Response resp = delegate.renameView(prefix, renameTableRequest, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_RENAME_VIEW,
eventMetadataFactory.create(),
@@ -734,7 +734,7 @@ public class IcebergRestCatalogEventServiceDelegator
SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
Namespace namespaceObj = decodeNamespace(namespace);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_REPLACE_VIEW,
eventMetadataFactory.create(),
@@ -746,7 +746,7 @@ public class IcebergRestCatalogEventServiceDelegator
Response resp =
delegate.replaceView(
prefix, namespace, view, commitViewRequest, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_REPLACE_VIEW,
eventMetadataFactory.create(),
@@ -766,7 +766,7 @@ public class IcebergRestCatalogEventServiceDelegator
RealmContext realmContext,
SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_COMMIT_TRANSACTION,
eventMetadataFactory.create(),
@@ -774,7 +774,7 @@ public class IcebergRestCatalogEventServiceDelegator
.put(EventAttributes.CATALOG_NAME, catalogName)
.put(EventAttributes.COMMIT_TRANSACTION_REQUEST, commitTransactionRequest)));
for (UpdateTableRequest req : commitTransactionRequest.tableChanges()) {
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_UPDATE_TABLE,
eventMetadataFactory.create(),
@@ -786,7 +786,7 @@ public class IcebergRestCatalogEventServiceDelegator
}
Response resp =
delegate.commitTransaction(prefix, commitTransactionRequest, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_COMMIT_TRANSACTION,
eventMetadataFactory.create(),
@@ -801,7 +801,7 @@ public class IcebergRestCatalogEventServiceDelegator
tableMetadataList != null && i < tableMetadataList.size()
? tableMetadataList.get(i)
: null;
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_UPDATE_TABLE,
eventMetadataFactory.create(),
@@ -837,7 +837,7 @@ public class IcebergRestCatalogEventServiceDelegator
SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
Namespace namespaceObj = decodeNamespace(namespace);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_SEND_NOTIFICATION,
eventMetadataFactory.create(),
@@ -849,7 +849,7 @@ public class IcebergRestCatalogEventServiceDelegator
Response resp =
delegate.sendNotification(
prefix, namespace, table, notificationRequest, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_SEND_NOTIFICATION,
eventMetadataFactory.create(),
diff --git a/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergRestConfigurationEventServiceDelegator.java b/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergRestConfigurationEventServiceDelegator.java
index 06c2382650..56b3c62ba9 100644
--- a/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergRestConfigurationEventServiceDelegator.java
+++ b/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergRestConfigurationEventServiceDelegator.java
@@ -32,9 +32,9 @@ import org.apache.polaris.service.catalog.api.IcebergRestConfigurationApiService
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)
@@ -42,16 +42,16 @@ public class IcebergRestConfigurationEventServiceDelegator
implements IcebergRestConfigurationApiService {
@Inject @Delegate IcebergCatalogAdapter delegate;
- @Inject PolarisEventListener polarisEventListener;
+ @Inject PolarisEventDispatcher polarisEventDispatcher;
@Inject PolarisEventMetadataFactory eventMetadataFactory;
@VisibleForTesting
public IcebergRestConfigurationEventServiceDelegator(
IcebergCatalogAdapter delegate,
- PolarisEventListener polarisEventListener,
+ PolarisEventDispatcher polarisEventDispatcher,
PolarisEventMetadataFactory eventMetadataFactory) {
this.delegate = delegate;
- this.polarisEventListener = polarisEventListener;
+ this.polarisEventDispatcher = polarisEventDispatcher;
this.eventMetadataFactory = eventMetadataFactory;
}
@@ -60,13 +60,13 @@ public class IcebergRestConfigurationEventServiceDelegator
@Override
public Response getConfig(
String warehouse, RealmContext realmContext, SecurityContext securityContext) {
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_GET_CONFIG,
eventMetadataFactory.create(),
new EventAttributeMap().put(EventAttributes.WAREHOUSE, warehouse)));
Response resp = delegate.getConfig(warehouse, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_GET_CONFIG,
eventMetadataFactory.create(),
diff --git a/runtime/service/src/main/java/org/apache/polaris/service/catalog/policy/CatalogPolicyEventServiceDelegator.java b/runtime/service/src/main/java/org/apache/polaris/service/catalog/policy/CatalogPolicyEventServiceDelegator.java
index 6a0682d876..7fe534aaf1 100644
--- a/runtime/service/src/main/java/org/apache/polaris/service/catalog/policy/CatalogPolicyEventServiceDelegator.java
+++ b/runtime/service/src/main/java/org/apache/polaris/service/catalog/policy/CatalogPolicyEventServiceDelegator.java
@@ -32,9 +32,9 @@ import org.apache.polaris.service.catalog.common.CatalogAdapter;
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;
import org.apache.polaris.service.types.AttachPolicyRequest;
import org.apache.polaris.service.types.CreatePolicyRequest;
import org.apache.polaris.service.types.DetachPolicyRequest;
@@ -48,7 +48,7 @@ public class CatalogPolicyEventServiceDelegator
implements PolarisCatalogPolicyApiService, CatalogAdapter {
@Inject @Delegate PolicyCatalogAdapter delegate;
- @Inject PolarisEventListener polarisEventListener;
+ @Inject PolarisEventDispatcher polarisEventDispatcher;
@Inject PolarisEventMetadataFactory eventMetadataFactory;
@Inject CatalogPrefixParser prefixParser;
@@ -60,7 +60,7 @@ public class CatalogPolicyEventServiceDelegator
RealmContext realmContext,
SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_CREATE_POLICY,
eventMetadataFactory.create(),
@@ -71,7 +71,7 @@ public class CatalogPolicyEventServiceDelegator
Response resp =
delegate.createPolicy(
prefix, namespace, createPolicyRequest, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_CREATE_POLICY,
eventMetadataFactory.create(),
@@ -92,7 +92,7 @@ public class CatalogPolicyEventServiceDelegator
RealmContext realmContext,
SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_LIST_POLICIES,
eventMetadataFactory.create(),
@@ -103,7 +103,7 @@ public class CatalogPolicyEventServiceDelegator
Response resp =
delegate.listPolicies(
prefix, namespace, pageToken, pageSize, policyType, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_LIST_POLICIES,
eventMetadataFactory.create(),
@@ -122,7 +122,7 @@ public class CatalogPolicyEventServiceDelegator
RealmContext realmContext,
SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_LOAD_POLICY,
eventMetadataFactory.create(),
@@ -132,7 +132,7 @@ public class CatalogPolicyEventServiceDelegator
.put(EventAttributes.POLICY_NAME, policyName)));
Response resp =
delegate.loadPolicy(prefix, namespace, policyName, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_LOAD_POLICY,
eventMetadataFactory.create(),
@@ -152,7 +152,7 @@ public class CatalogPolicyEventServiceDelegator
RealmContext realmContext,
SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_UPDATE_POLICY,
eventMetadataFactory.create(),
@@ -164,7 +164,7 @@ public class CatalogPolicyEventServiceDelegator
Response resp =
delegate.updatePolicy(
prefix, namespace, policyName, updatePolicyRequest, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_UPDATE_POLICY,
eventMetadataFactory.create(),
@@ -184,7 +184,7 @@ public class CatalogPolicyEventServiceDelegator
RealmContext realmContext,
SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_DROP_POLICY,
eventMetadataFactory.create(),
@@ -196,7 +196,7 @@ public class CatalogPolicyEventServiceDelegator
Response resp =
delegate.dropPolicy(
prefix, namespace, policyName, detachAll, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_DROP_POLICY,
eventMetadataFactory.create(),
@@ -217,7 +217,7 @@ public class CatalogPolicyEventServiceDelegator
RealmContext realmContext,
SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_ATTACH_POLICY,
eventMetadataFactory.create(),
@@ -229,7 +229,7 @@ public class CatalogPolicyEventServiceDelegator
Response resp =
delegate.attachPolicy(
prefix, namespace, policyName, attachPolicyRequest, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_ATTACH_POLICY,
eventMetadataFactory.create(),
@@ -250,7 +250,7 @@ public class CatalogPolicyEventServiceDelegator
RealmContext realmContext,
SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_DETACH_POLICY,
eventMetadataFactory.create(),
@@ -262,7 +262,7 @@ public class CatalogPolicyEventServiceDelegator
Response resp =
delegate.detachPolicy(
prefix, namespace, policyName, detachPolicyRequest, realmContext, securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_DETACH_POLICY,
eventMetadataFactory.create(),
@@ -285,7 +285,7 @@ public class CatalogPolicyEventServiceDelegator
RealmContext realmContext,
SecurityContext securityContext) {
String catalogName = prefixParser.prefixToCatalogName(prefix);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_GET_APPLICABLE_POLICIES,
eventMetadataFactory.create(),
@@ -304,7 +304,7 @@ public class CatalogPolicyEventServiceDelegator
policyType,
realmContext,
securityContext);
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_GET_APPLICABLE_POLICIES,
eventMetadataFactory.create(),
diff --git a/runtime/service/src/main/java/org/apache/polaris/service/config/ServiceProducers.java b/runtime/service/src/main/java/org/apache/polaris/service/config/ServiceProducers.java
index a741b0aff2..e25c01a6cc 100644
--- a/runtime/service/src/main/java/org/apache/polaris/service/config/ServiceProducers.java
+++ b/runtime/service/src/main/java/org/apache/polaris/service/config/ServiceProducers.java
@@ -74,8 +74,6 @@ import org.apache.polaris.service.catalog.io.FileIOFactory;
import org.apache.polaris.service.context.RealmContextConfiguration;
import org.apache.polaris.service.context.RealmContextResolver;
import org.apache.polaris.service.credentials.PolarisCredentialManagerConfiguration;
-import org.apache.polaris.service.events.PolarisEventListenerConfiguration;
-import org.apache.polaris.service.events.listeners.PolarisEventListener;
import org.apache.polaris.service.persistence.PersistenceConfiguration;
import org.apache.polaris.service.ratelimiter.RateLimiter;
import org.apache.polaris.service.ratelimiter.RateLimiterFilterConfiguration;
@@ -215,14 +213,6 @@ public class ServiceProducers {
return fileIOFactories.select(Identifier.Literal.of(config.type())).get();
}
- @Produces
- @Singleton // used in instanceof checks
- public PolarisEventListener polarisEventListener(
- PolarisEventListenerConfiguration config,
- @Any Instance<PolarisEventListener> polarisEventListeners) {
- return polarisEventListeners.select(Identifier.Literal.of(config.type())).get();
- }
-
@Produces
@Singleton // used in instanceof checks
public MetaStoreManagerFactory metaStoreManagerFactory(
diff --git a/runtime/service/src/main/java/org/apache/polaris/service/context/catalog/PolarisCallContextCatalogFactory.java b/runtime/service/src/main/java/org/apache/polaris/service/context/catalog/PolarisCallContextCatalogFactory.java
index 960967d8f0..8666c3f900 100644
--- a/runtime/service/src/main/java/org/apache/polaris/service/context/catalog/PolarisCallContextCatalogFactory.java
+++ b/runtime/service/src/main/java/org/apache/polaris/service/context/catalog/PolarisCallContextCatalogFactory.java
@@ -34,8 +34,8 @@ import org.apache.polaris.core.persistence.resolver.ResolverFactory;
import org.apache.polaris.service.catalog.iceberg.IcebergCatalog;
import org.apache.polaris.service.catalog.io.FileIOFactory;
import org.apache.polaris.service.catalog.io.StorageAccessConfigProvider;
+import org.apache.polaris.service.events.PolarisEventDispatcher;
import org.apache.polaris.service.events.PolarisEventMetadataFactory;
-import org.apache.polaris.service.events.listeners.PolarisEventListener;
import org.apache.polaris.service.task.TaskExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -50,7 +50,7 @@ public class PolarisCallContextCatalogFactory implements CallContextCatalogFacto
private final StorageAccessConfigProvider storageAccessConfigProvider;
private final FileIOFactory fileIOFactory;
private final ResolverFactory resolverFactory;
- private final PolarisEventListener polarisEventListener;
+ private final PolarisEventDispatcher polarisEventDispatcher;
private final PolarisEventMetadataFactory eventMetadataFactory;
private final PolarisMetaStoreManager metaStoreManager;
private final CallContext callContext;
@@ -63,7 +63,7 @@ public class PolarisCallContextCatalogFactory implements CallContextCatalogFacto
TaskExecutor taskExecutor,
StorageAccessConfigProvider storageAccessConfigProvider,
FileIOFactory fileIOFactory,
- PolarisEventListener polarisEventListener,
+ PolarisEventDispatcher polarisEventDispatcher,
PolarisEventMetadataFactory eventMetadataFactory,
PolarisMetaStoreManager metaStoreManager,
CallContext callContext,
@@ -73,7 +73,7 @@ public class PolarisCallContextCatalogFactory implements CallContextCatalogFacto
this.taskExecutor = taskExecutor;
this.storageAccessConfigProvider = storageAccessConfigProvider;
this.fileIOFactory = fileIOFactory;
- this.polarisEventListener = polarisEventListener;
+ this.polarisEventDispatcher = polarisEventDispatcher;
this.eventMetadataFactory = eventMetadataFactory;
this.metaStoreManager = metaStoreManager;
this.callContext = callContext;
@@ -100,7 +100,7 @@ public class PolarisCallContextCatalogFactory implements CallContextCatalogFacto
taskExecutor,
storageAccessConfigProvider,
fileIOFactory,
- polarisEventListener,
+ polarisEventDispatcher,
eventMetadataFactory);
Map<String, String> catalogProperties = new HashMap<>(catalog.getPropertiesAsMap());
diff --git a/runtime/service/src/main/java/org/apache/polaris/service/events/listeners/NoOpPolarisEventListener.java b/runtime/service/src/main/java/org/apache/polaris/service/events/EventBusConfigurer.java
similarity index 64%
copy from runtime/service/src/main/java/org/apache/polaris/service/events/listeners/NoOpPolarisEventListener.java
copy to runtime/service/src/main/java/org/apache/polaris/service/events/EventBusConfigurer.java
index bfef6e96ec..c822599008 100644
--- a/runtime/service/src/main/java/org/apache/polaris/service/events/listeners/NoOpPolarisEventListener.java
+++ b/runtime/service/src/main/java/org/apache/polaris/service/events/EventBusConfigurer.java
@@ -16,12 +16,20 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.polaris.service.events.listeners;
-import io.smallrye.common.annotation.Identifier;
+package org.apache.polaris.service.events;
+
+import io.quarkus.runtime.StartupEvent;
+import io.quarkus.vertx.LocalEventBusCodec;
+import io.vertx.core.eventbus.EventBus;
import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.enterprise.event.Observes;
-/** Event listener that does nothing. */
@ApplicationScoped
-@Identifier("no-op")
-public class NoOpPolarisEventListener implements PolarisEventListener {}
+public class EventBusConfigurer {
+ private static final String LOCAL_CODEC_NAME = "polaris-events-local";
+
+ void configureEventBus(@Observes StartupEvent ev, EventBus eventBus) {
+ eventBus.registerDefaultCodec(PolarisEvent.class, new LocalEventBusCodec<>(LOCAL_CODEC_NAME));
+ }
+}
diff --git a/runtime/service/src/main/java/org/apache/polaris/service/events/listeners/NoOpPolarisEventListener.java b/runtime/service/src/main/java/org/apache/polaris/service/events/FallbackEventListenerTypeConfigInterceptor.java
similarity index 68%
copy from runtime/service/src/main/java/org/apache/polaris/service/events/listeners/NoOpPolarisEventListener.java
copy to runtime/service/src/main/java/org/apache/polaris/service/events/FallbackEventListenerTypeConfigInterceptor.java
index bfef6e96ec..abadb7e6d9 100644
--- a/runtime/service/src/main/java/org/apache/polaris/service/events/listeners/NoOpPolarisEventListener.java
+++ b/runtime/service/src/main/java/org/apache/polaris/service/events/FallbackEventListenerTypeConfigInterceptor.java
@@ -16,12 +16,14 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.polaris.service.events.listeners;
-import io.smallrye.common.annotation.Identifier;
-import jakarta.enterprise.context.ApplicationScoped;
+package org.apache.polaris.service.events;
-/** Event listener that does nothing. */
-@ApplicationScoped
-@Identifier("no-op")
-public class NoOpPolarisEventListener implements PolarisEventListener {}
+import io.smallrye.config.FallbackConfigSourceInterceptor;
+
+public class FallbackEventListenerTypeConfigInterceptor extends FallbackConfigSourceInterceptor {
+ public FallbackEventListenerTypeConfigInterceptor() {
+ super(
+ name -> "polaris.event-listener.types".equals(name) ? "polaris.event-listener.type" : name);
+ }
+}
diff --git a/runtime/service/src/main/java/org/apache/polaris/service/events/listeners/NoOpPolarisEventListener.java b/runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventDispatcher.java
similarity index 72%
copy from runtime/service/src/main/java/org/apache/polaris/service/events/listeners/NoOpPolarisEventListener.java
copy to runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventDispatcher.java
index bfef6e96ec..f07f72abca 100644
--- a/runtime/service/src/main/java/org/apache/polaris/service/events/listeners/NoOpPolarisEventListener.java
+++ b/runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventDispatcher.java
@@ -16,12 +16,9 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.polaris.service.events.listeners;
-import io.smallrye.common.annotation.Identifier;
-import jakarta.enterprise.context.ApplicationScoped;
+package org.apache.polaris.service.events;
-/** Event listener that does nothing. */
-@ApplicationScoped
-@Identifier("no-op")
-public class NoOpPolarisEventListener implements PolarisEventListener {}
+public interface PolarisEventDispatcher {
+ void dispatch(PolarisEvent event);
+}
diff --git a/runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListenerConfiguration.java b/runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListenerConfiguration.java
index 56c12eac03..ec38fe1d79 100644
--- a/runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListenerConfiguration.java
+++ b/runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListenerConfiguration.java
@@ -20,6 +20,8 @@ package org.apache.polaris.service.events;
import io.quarkus.runtime.annotations.StaticInitSafe;
import io.smallrye.config.ConfigMapping;
+import java.util.Optional;
+import java.util.Set;
import org.apache.polaris.service.events.listeners.PolarisEventListener;
@StaticInitSafe
@@ -28,6 +30,15 @@ public interface PolarisEventListenerConfiguration {
/**
* The type of the event listener to use. Must be a registered {@link PolarisEventListener}
* identifier.
+ *
+ * @deprecated since 1.5.0, use 'polaris.event-listener.types' instead
*/
- String type();
+ @Deprecated(since = "1.5.0", forRemoval = true)
+ Optional<String> type();
+
+ /**
+ * Comma separated list of event listers, each item must be a registered {@link
+ * PolarisEventListener} identifier.
+ */
+ Optional<Set<String>> types();
}
diff --git a/runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java b/runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java
new file mode 100644
index 0000000000..06c7fb31cd
--- /dev/null
+++ b/runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.polaris.service.events;
+
+import static org.apache.polaris.service.events.PolarisServiceBusEventDispatcher.POLARIS_EVENT_CHANNEL;
+
+import io.quarkus.runtime.StartupEvent;
+import io.smallrye.common.annotation.Identifier;
+import io.vertx.core.Handler;
+import io.vertx.core.eventbus.EventBus;
+import io.vertx.core.eventbus.Message;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.enterprise.event.Observes;
+import jakarta.enterprise.inject.Any;
+import jakarta.enterprise.inject.Instance;
+import jakarta.inject.Inject;
+import java.util.HashSet;
+import java.util.Set;
+import org.apache.polaris.service.events.listeners.PolarisEventListener;
+
+@ApplicationScoped
+public class PolarisEventListeners {
+ @Inject EventBus eventBus;
+ @Inject @Any Instance<PolarisEventListener> eventListeners;
+ @Inject PolarisEventListenerConfiguration configuration;
+
+ public void onStartup(@Observes StartupEvent event) {
+ Set<String> listenerTypeSet = configuration.types().orElseGet(HashSet::new);
+ for (String enabledEventListener : listenerTypeSet) {
+ PolarisEventListener listener =
+ eventListeners.select(Identifier.Literal.of(enabledEventListener)).get();
+ Handler<Message<PolarisEvent>> handler = e -> listener.onEvent(e.body());
+ eventBus.localConsumer(POLARIS_EVENT_CHANNEL, handler);
+ }
+ }
+}
diff --git a/runtime/service/src/main/java/org/apache/polaris/service/events/listeners/NoOpPolarisEventListener.java b/runtime/service/src/main/java/org/apache/polaris/service/events/PolarisServiceBusEventDispatcher.java
similarity index 53%
copy from runtime/service/src/main/java/org/apache/polaris/service/events/listeners/NoOpPolarisEventListener.java
copy to runtime/service/src/main/java/org/apache/polaris/service/events/PolarisServiceBusEventDispatcher.java
index bfef6e96ec..91d13544c1 100644
--- a/runtime/service/src/main/java/org/apache/polaris/service/events/listeners/NoOpPolarisEventListener.java
+++ b/runtime/service/src/main/java/org/apache/polaris/service/events/PolarisServiceBusEventDispatcher.java
@@ -16,12 +16,30 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.polaris.service.events.listeners;
-import io.smallrye.common.annotation.Identifier;
+package org.apache.polaris.service.events;
+
+import io.quarkus.arc.DefaultBean;
+import io.vertx.core.eventbus.EventBus;
import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
-/** Event listener that does nothing. */
+/** An event dispatcher that publishes events to the Vert.x event bus. */
@ApplicationScoped
-@Identifier("no-op")
-public class NoOpPolarisEventListener implements PolarisEventListener {}
+@DefaultBean
+public class PolarisServiceBusEventDispatcher implements PolarisEventDispatcher {
+ /** The address of the event bus channel where Polaris service events are published. */
+ public static final String POLARIS_EVENT_CHANNEL = "polaris-events";
+
+ @Inject EventBus eventBus;
+
+ /**
+ * Dispatches the given Polaris service event to the Vert.x event bus.
+ *
+ * @param event The Polaris service event to dispatch.
+ */
+ @Override
+ public void dispatch(PolarisEvent event) {
+ eventBus.publish(POLARIS_EVENT_CHANNEL, event);
+ }
+}
diff --git a/runtime/service/src/main/java/org/apache/polaris/service/ratelimiter/RateLimiterFilter.java b/runtime/service/src/main/java/org/apache/polaris/service/ratelimiter/RateLimiterFilter.java
index 2572c7277e..add802ad2a 100644
--- a/runtime/service/src/main/java/org/apache/polaris/service/ratelimiter/RateLimiterFilter.java
+++ b/runtime/service/src/main/java/org/apache/polaris/service/ratelimiter/RateLimiterFilter.java
@@ -31,9 +31,9 @@ import org.apache.polaris.service.config.FilterPriorities;
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;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -46,16 +46,16 @@ public class RateLimiterFilter implements ContainerRequestFilter {
private static final Logger LOGGER = LoggerFactory.getLogger(RateLimiterFilter.class);
private final RateLimiter rateLimiter;
- private final PolarisEventListener polarisEventListener;
+ private final PolarisEventDispatcher polarisEventDispatcher;
private final PolarisEventMetadataFactory eventMetadataFactory;
@Inject
public RateLimiterFilter(
RateLimiter rateLimiter,
- PolarisEventListener polarisEventListener,
+ PolarisEventDispatcher polarisEventDispatcher,
PolarisEventMetadataFactory eventMetadataFactory) {
this.rateLimiter = rateLimiter;
- this.polarisEventListener = polarisEventListener;
+ this.polarisEventDispatcher = polarisEventDispatcher;
this.eventMetadataFactory = eventMetadataFactory;
}
@@ -63,7 +63,7 @@ public class RateLimiterFilter implements ContainerRequestFilter {
@Override
public void filter(ContainerRequestContext ctx) throws IOException {
if (!rateLimiter.canProceed()) {
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_LIMIT_REQUEST_RATE,
eventMetadataFactory.create(),
diff --git a/runtime/service/src/main/java/org/apache/polaris/service/task/TaskExecutorImpl.java b/runtime/service/src/main/java/org/apache/polaris/service/task/TaskExecutorImpl.java
index 2808857fee..5d0f4c6b33 100644
--- a/runtime/service/src/main/java/org/apache/polaris/service/task/TaskExecutorImpl.java
+++ b/runtime/service/src/main/java/org/apache/polaris/service/task/TaskExecutorImpl.java
@@ -53,10 +53,10 @@ import org.apache.polaris.service.context.catalog.RealmContextHolder;
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.PolarisEventMetadata;
import org.apache.polaris.service.events.PolarisEventMetadataFactory;
import org.apache.polaris.service.events.PolarisEventType;
-import org.apache.polaris.service.events.listeners.PolarisEventListener;
import org.apache.polaris.service.tracing.TracingFilter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -79,7 +79,7 @@ public class TaskExecutorImpl implements TaskExecutor {
private final PolarisPrincipal polarisPrincipal;
private final List<TaskHandler> taskHandlers = new CopyOnWriteArrayList<>();
private final Optional<TriConsumer<Long, Boolean, Throwable>> errorHandler;
- private final PolarisEventListener polarisEventListener;
+ private final PolarisEventDispatcher polarisEventDispatcher;
private final PolarisEventMetadataFactory eventMetadataFactory;
@Nullable private final Tracer tracer;
@@ -97,7 +97,7 @@ public class TaskExecutorImpl implements TaskExecutor {
MetaStoreManagerFactory metaStoreManagerFactory,
TaskFileIOSupplier fileIOSupplier,
RealmContextHolder realmContextHolder,
- PolarisEventListener polarisEventListener,
+ PolarisEventDispatcher polarisEventDispatcher,
PolarisEventMetadataFactory eventMetadataFactory,
@Nullable Tracer tracer,
PolarisPrincipalHolder polarisPrincipalHolder,
@@ -107,7 +107,7 @@ public class TaskExecutorImpl implements TaskExecutor {
this.metaStoreManagerFactory = metaStoreManagerFactory;
this.fileIOSupplier = fileIOSupplier;
this.realmContextHolder = realmContextHolder;
- this.polarisEventListener = polarisEventListener;
+ this.polarisEventDispatcher = polarisEventDispatcher;
this.eventMetadataFactory = eventMetadataFactory;
this.tracer = tracer;
this.polarisPrincipalHolder = polarisPrincipalHolder;
@@ -198,7 +198,7 @@ public class TaskExecutorImpl implements TaskExecutor {
protected void handleTask(
long taskEntityId, CallContext ctx, PolarisEventMetadata eventMetadata, int attempt) {
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.BEFORE_ATTEMPT_TASK,
eventMetadataFactory.copy(eventMetadata),
@@ -247,7 +247,7 @@ public class TaskExecutorImpl implements TaskExecutor {
.log("Unable to execute async task");
}
} finally {
- polarisEventListener.onEvent(
+ polarisEventDispatcher.dispatch(
new PolarisEvent(
PolarisEventType.AFTER_ATTEMPT_TASK,
eventMetadataFactory.copy(eventMetadata),
diff --git a/site/content/in-dev/unreleased/configuration/config-sections/smallrye-polaris_event_listener.md b/runtime/service/src/main/resources/META-INF/services/io.smallrye.config.ConfigSourceInterceptor
similarity index 70%
copy from site/content/in-dev/unreleased/configuration/config-sections/smallrye-polaris_event_listener.md
copy to runtime/service/src/main/resources/META-INF/services/io.smallrye.config.ConfigSourceInterceptor
index ef266451e8..2ec98e2df2 100644
--- a/site/content/in-dev/unreleased/configuration/config-sections/smallrye-polaris_event_listener.md
+++ b/runtime/service/src/main/resources/META-INF/services/io.smallrye.config.ConfigSourceInterceptor
@@ -1,4 +1,3 @@
----
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
@@ -17,12 +16,5 @@
# specific language governing permissions and limitations
# under the License.
#
-title: smallrye-polaris_event_listener
-build:
- list: never
- render: never
----
-| Property | Default Value | Type | Description |
-|----------|---------------|------|-------------|
-| `polaris.event-listener.type` | | `string` | The type of the event listener to use. Must be a registered (`PolarisEventListener`) identifier. |
+org.apache.polaris.service.events.FallbackEventListenerTypeConfigInterceptor
\ No newline at end of file
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/admin/PolarisAuthzTestBase.java b/runtime/service/src/test/java/org/apache/polaris/service/admin/PolarisAuthzTestBase.java
index 1e387b5f8d..cbdec6ca59 100644
--- a/runtime/service/src/test/java/org/apache/polaris/service/admin/PolarisAuthzTestBase.java
+++ b/runtime/service/src/test/java/org/apache/polaris/service/admin/PolarisAuthzTestBase.java
@@ -84,8 +84,8 @@ import org.apache.polaris.service.catalog.policy.PolicyCatalog;
import org.apache.polaris.service.config.ReservedProperties;
import org.apache.polaris.service.context.catalog.PolarisCallContextCatalogFactory;
import org.apache.polaris.service.context.catalog.RealmContextHolder;
+import org.apache.polaris.service.events.PolarisEventDispatcher;
import org.apache.polaris.service.events.PolarisEventMetadataFactory;
-import org.apache.polaris.service.events.listeners.PolarisEventListener;
import org.apache.polaris.service.storage.PolarisStorageIntegrationProviderImpl;
import org.apache.polaris.service.task.TaskExecutor;
import org.apache.polaris.service.types.PolicyIdentifier;
@@ -186,7 +186,6 @@ public abstract class PolarisAuthzTestBase {
@Inject protected ServiceIdentityProvider serviceIdentityProvider;
@Inject protected PolarisDiagnostics diagServices;
@Inject protected FileIOFactory fileIOFactory;
- @Inject protected PolarisEventListener polarisEventListener;
@Inject protected PolarisEventMetadataFactory eventMetadataFactory;
@Inject protected StorageCredentialCache storageCredentialCache;
@Inject protected ResolverFactory resolverFactory;
@@ -196,6 +195,7 @@ public abstract class PolarisAuthzTestBase {
@Inject protected CallContext callContext;
@Inject protected RealmConfig realmConfig;
@Inject protected RealmContextHolder realmContextHolder;
+ @Inject protected PolarisEventDispatcher polarisEventDispatcher;
protected IcebergCatalog baseCatalog;
protected PolarisGenericTableCatalog genericTableCatalog;
@@ -486,7 +486,7 @@ public abstract class PolarisAuthzTestBase {
Mockito.mock(),
storageAccessConfigProvider,
fileIOFactory,
- polarisEventListener,
+ polarisEventDispatcher,
eventMetadataFactory);
this.baseCatalog.initialize(
CATALOG_NAME,
@@ -515,7 +515,7 @@ public abstract class PolarisAuthzTestBase {
TaskExecutor taskExecutor,
StorageAccessConfigProvider accessConfigProvider,
FileIOFactory fileIOFactory,
- PolarisEventListener polarisEventListener,
+ PolarisEventDispatcher polarisEventDispatcher,
PolarisEventMetadataFactory eventMetadataFactory,
PolarisMetaStoreManager metaStoreManager,
CallContext callContext,
@@ -526,7 +526,7 @@ public abstract class PolarisAuthzTestBase {
taskExecutor,
accessConfigProvider,
fileIOFactory,
- polarisEventListener,
+ polarisEventDispatcher,
eventMetadataFactory,
metaStoreManager,
callContext,
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/catalog/generic/AbstractPolarisGenericTableCatalogTest.java b/runtime/service/src/test/java/org/apache/polaris/service/catalog/generic/AbstractPolarisGenericTableCatalogTest.java
index 9e1f3a0d36..4372cc2f00 100644
--- a/runtime/service/src/test/java/org/apache/polaris/service/catalog/generic/AbstractPolarisGenericTableCatalogTest.java
+++ b/runtime/service/src/test/java/org/apache/polaris/service/catalog/generic/AbstractPolarisGenericTableCatalogTest.java
@@ -69,7 +69,7 @@ import org.apache.polaris.service.catalog.io.StorageAccessConfigProvider;
import org.apache.polaris.service.config.ReservedProperties;
import org.apache.polaris.service.context.catalog.PolarisPrincipalHolder;
import org.apache.polaris.service.events.PolarisEventMetadataFactory;
-import org.apache.polaris.service.events.listeners.NoOpPolarisEventListener;
+import org.apache.polaris.service.events.listeners.InMemoryEventCollector;
import org.apache.polaris.service.storage.PolarisStorageIntegrationProviderImpl;
import org.apache.polaris.service.task.TaskExecutor;
import org.assertj.core.api.Assertions;
@@ -234,7 +234,7 @@ public abstract class AbstractPolarisGenericTableCatalogTest {
taskExecutor,
storageAccessConfigProvider,
fileIOFactory,
- new NoOpPolarisEventListener(),
+ new InMemoryEventCollector(),
eventMetadataFactory);
this.icebergCatalog.initialize(
CATALOG_NAME,
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/AbstractIcebergCatalogHandlerAuthzTest.java b/runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/AbstractIcebergCatalogHandlerAuthzTest.java
index ee008865a8..7a9048449a 100644
--- a/runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/AbstractIcebergCatalogHandlerAuthzTest.java
+++ b/runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/AbstractIcebergCatalogHandlerAuthzTest.java
@@ -1457,7 +1457,7 @@ public abstract class AbstractIcebergCatalogHandlerAuthzTest extends PolarisAuth
Mockito.mock(),
storageAccessConfigProvider,
fileIOFactory,
- polarisEventListener,
+ polarisEventDispatcher,
eventMetadataFactory,
metaStoreManager,
callContext,
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/AbstractIcebergCatalogTest.java b/runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/AbstractIcebergCatalogTest.java
index 46624c297a..0b6481a95c 100644
--- a/runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/AbstractIcebergCatalogTest.java
+++ b/runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/AbstractIcebergCatalogTest.java
@@ -38,6 +38,7 @@ import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterables;
import com.google.common.collect.Streams;
import io.quarkus.test.junit.QuarkusMock;
+import io.smallrye.common.annotation.Identifier;
import jakarta.annotation.Nonnull;
import jakarta.inject.Inject;
import java.io.IOException;
@@ -140,6 +141,7 @@ import org.apache.polaris.service.catalog.io.StorageAccessConfigProvider;
import org.apache.polaris.service.context.catalog.RealmContextHolder;
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;
@@ -203,6 +205,7 @@ public abstract class AbstractIcebergCatalogTest extends CatalogTests<IcebergCat
.put("polaris.features.\"LIST_PAGINATION_ENABLED\"", "true")
.put("polaris.behavior-changes.\"ALLOW_NAMESPACE_CUSTOM_LOCATION\"", "true")
.put("polaris.test.rootAugmentor.enabled", "true")
+ .put("polaris.event-listener.types", "test")
.build();
}
}
@@ -233,7 +236,11 @@ public abstract class AbstractIcebergCatalogTest extends CatalogTests<IcebergCat
@Inject PolarisStorageIntegrationProvider storageIntegrationProvider;
@Inject ServiceIdentityProvider serviceIdentityProvider;
@Inject PolarisDiagnostics diagServices;
- @Inject PolarisEventListener polarisEventListener;
+
+ @Inject
+ @Identifier("test")
+ PolarisEventListener polarisEventListener;
+
@Inject PolarisEventMetadataFactory eventMetadataFactory;
@Inject PolarisMetaStoreManager metaStoreManager;
@Inject CallContext callContext;
@@ -246,6 +253,7 @@ public abstract class AbstractIcebergCatalogTest extends CatalogTests<IcebergCat
@Inject PolarisPrincipal authenticatedRoot;
@Inject PolarisAdminService adminService;
@Inject ResolverFactory resolverFactory;
+ @Inject PolarisEventDispatcher polarisEventDispatcher;
private IcebergCatalog catalog;
private String realmName;
@@ -407,7 +415,7 @@ public abstract class AbstractIcebergCatalogTest extends CatalogTests<IcebergCat
taskExecutor,
storageAccessConfigProvider,
fileIOFactory,
- polarisEventListener,
+ polarisEventDispatcher,
eventMetadataFactory);
}
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/AbstractIcebergCatalogViewTest.java b/runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/AbstractIcebergCatalogViewTest.java
index a6fc777b0c..1d6299dc46 100644
--- a/runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/AbstractIcebergCatalogViewTest.java
+++ b/runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/AbstractIcebergCatalogViewTest.java
@@ -20,6 +20,7 @@ package org.apache.polaris.service.catalog.iceberg;
import com.google.common.collect.ImmutableMap;
import io.quarkus.test.junit.QuarkusMock;
+import io.smallrye.common.annotation.Identifier;
import jakarta.inject.Inject;
import java.io.IOException;
import java.lang.reflect.Field;
@@ -60,6 +61,7 @@ import org.apache.polaris.service.catalog.io.StorageAccessConfigProvider;
import org.apache.polaris.service.config.ReservedProperties;
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;
@@ -90,6 +92,7 @@ public abstract class AbstractIcebergCatalogViewTest extends ViewCatalogTests<Ic
.put("polaris.features.\"ALLOW_WILDCARD_LOCATION\"", "true")
.put("polaris.features.\"SKIP_CREDENTIAL_SUBSCOPING_INDIRECTION\"", "true")
.put("polaris.features.\"LIST_PAGINATION_ENABLED\"", "true")
+ .put("polaris.event-listener.types", "test")
.build();
}
}
@@ -107,7 +110,12 @@ public abstract class AbstractIcebergCatalogViewTest extends ViewCatalogTests<Ic
@Inject ServiceIdentityProvider serviceIdentityProvider;
@Inject StorageCredentialCache storageCredentialCache;
@Inject PolarisDiagnostics diagServices;
- @Inject PolarisEventListener polarisEventListener;
+
+ @Inject
+ @Identifier("test")
+ PolarisEventListener polarisEventListener;
+
+ @Inject PolarisEventDispatcher polarisEventDispatcher;
@Inject PolarisEventMetadataFactory eventMetadataFactory;
@Inject ResolverFactory resolverFactory;
@Inject ResolutionManifestFactory resolutionManifestFactory;
@@ -207,7 +215,7 @@ public abstract class AbstractIcebergCatalogViewTest extends ViewCatalogTests<Ic
Mockito.mock(),
storageAccessConfigProvider,
fileIOFactory,
- polarisEventListener,
+ polarisEventDispatcher,
eventMetadataFactory);
Map<String, String> properties =
ImmutableMap.<String, String>builder()
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/CommitTransactionEventTest.java b/runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/CommitTransactionEventTest.java
index 46822e28e5..8c42052b73 100644
--- a/runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/CommitTransactionEventTest.java
+++ b/runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/CommitTransactionEventTest.java
@@ -45,7 +45,7 @@ import org.apache.polaris.service.TestServices;
import org.apache.polaris.service.events.EventAttributes;
import org.apache.polaris.service.events.PolarisEvent;
import org.apache.polaris.service.events.PolarisEventType;
-import org.apache.polaris.service.events.listeners.TestPolarisEventListener;
+import org.apache.polaris.service.events.listeners.InMemoryEventCollector;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
@@ -76,17 +76,19 @@ public class CommitTransactionEventTest {
// Verify that all (Before/After)CommitTransaction and (Before/After)UpdateTable events were
// emitted
- TestPolarisEventListener testEventListener =
- (TestPolarisEventListener) testServices.polarisEventListener();
- assertThat(testEventListener.getLatest(PolarisEventType.BEFORE_COMMIT_TRANSACTION)).isNotNull();
+ InMemoryEventCollector testPolarisEventDispatcher =
+ (InMemoryEventCollector) testServices.polarisEventDispatcher();
+ assertThat(testPolarisEventDispatcher.getLatest(PolarisEventType.BEFORE_COMMIT_TRANSACTION))
+ .isNotNull();
PolarisEvent beforeUpdateEvent =
- testEventListener.getLatest(PolarisEventType.BEFORE_UPDATE_TABLE);
+ testPolarisEventDispatcher.getLatest(PolarisEventType.BEFORE_UPDATE_TABLE);
assertThat(beforeUpdateEvent.attributes().getRequired(EventAttributes.TABLE_NAME))
.isEqualTo(table2Name);
- assertThat(testEventListener.getLatest(PolarisEventType.AFTER_COMMIT_TRANSACTION)).isNotNull();
+ assertThat(testPolarisEventDispatcher.getLatest(PolarisEventType.AFTER_COMMIT_TRANSACTION))
+ .isNotNull();
PolarisEvent afterUpdateEvent =
- testEventListener.getLatest(PolarisEventType.AFTER_UPDATE_TABLE);
+ testPolarisEventDispatcher.getLatest(PolarisEventType.AFTER_UPDATE_TABLE);
assertThat(afterUpdateEvent.attributes().getRequired(EventAttributes.TABLE_NAME))
.isEqualTo(table2Name);
}
@@ -101,20 +103,23 @@ public class CommitTransactionEventTest {
executeTransactionTest(true, table3Name, table4Name, testServices);
// Verify that all (Before)CommitTable events were emitted
- TestPolarisEventListener testEventListener =
- (TestPolarisEventListener) testServices.polarisEventListener();
+ InMemoryEventCollector testPolarisEventDispatcher =
+ (InMemoryEventCollector) testServices.polarisEventDispatcher();
// Verify that all BeforeCommitTransaction and BeforeUpdateTable events were emitted,
// and that the AfterCommitTransaction and AfterUpdateTable events were not emitted
- assertThat(testEventListener.getLatest(PolarisEventType.BEFORE_COMMIT_TRANSACTION)).isNotNull();
+ assertThat(testPolarisEventDispatcher.getLatest(PolarisEventType.BEFORE_COMMIT_TRANSACTION))
+ .isNotNull();
PolarisEvent beforeUpdateEvent =
- testEventListener.getLatest(PolarisEventType.BEFORE_UPDATE_TABLE);
+ testPolarisEventDispatcher.getLatest(PolarisEventType.BEFORE_UPDATE_TABLE);
assertThat(beforeUpdateEvent.attributes().getRequired(EventAttributes.TABLE_NAME))
.isEqualTo(table4Name);
- assertThatThrownBy(() -> testEventListener.getLatest(PolarisEventType.AFTER_COMMIT_TRANSACTION))
+ assertThatThrownBy(
+ () -> testPolarisEventDispatcher.getLatest(PolarisEventType.AFTER_COMMIT_TRANSACTION))
.isInstanceOf(IllegalStateException.class);
- assertThatThrownBy(() -> testEventListener.getLatest(PolarisEventType.AFTER_UPDATE_TABLE))
+ assertThatThrownBy(
+ () -> testPolarisEventDispatcher.getLatest(PolarisEventType.AFTER_UPDATE_TABLE))
.isInstanceOf(IllegalStateException.class);
}
@@ -127,12 +132,12 @@ public class CommitTransactionEventTest {
String table2Name = "test-table-6";
executeTransactionTest(false, table1Name, table2Name, testServices);
- TestPolarisEventListener testEventListener =
- (TestPolarisEventListener) testServices.polarisEventListener();
+ InMemoryEventCollector testPolarisEventDispatcher =
+ (InMemoryEventCollector) testServices.polarisEventDispatcher();
// Verify that AfterUpdateTable events contain LoadTableResponse objects
PolarisEvent afterUpdateTableEvent =
- testEventListener.getLatest(PolarisEventType.AFTER_UPDATE_TABLE);
+ testPolarisEventDispatcher.getLatest(PolarisEventType.AFTER_UPDATE_TABLE);
// Verify second table's LoadTableResponse
assertThat(afterUpdateTableEvent.attributes().getRequired(EventAttributes.TABLE_NAME))
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/catalog/io/FileIOFactoryTest.java b/runtime/service/src/test/java/org/apache/polaris/service/catalog/io/FileIOFactoryTest.java
index e0d2737219..44f5647c74 100644
--- a/runtime/service/src/test/java/org/apache/polaris/service/catalog/io/FileIOFactoryTest.java
+++ b/runtime/service/src/test/java/org/apache/polaris/service/catalog/io/FileIOFactoryTest.java
@@ -219,7 +219,7 @@ public class FileIOFactoryTest {
services.taskExecutor(),
services.storageAccessConfigProvider(),
services.fileIOFactory(),
- services.polarisEventListener(),
+ services.polarisEventDispatcher(),
services.eventMetadataFactory());
polarisCatalog.initialize(
CATALOG_NAME,
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/catalog/policy/AbstractPolicyCatalogTest.java b/runtime/service/src/test/java/org/apache/polaris/service/catalog/policy/AbstractPolicyCatalogTest.java
index e8c2d4699d..fe37f8c600 100644
--- a/runtime/service/src/test/java/org/apache/polaris/service/catalog/policy/AbstractPolicyCatalogTest.java
+++ b/runtime/service/src/test/java/org/apache/polaris/service/catalog/policy/AbstractPolicyCatalogTest.java
@@ -81,7 +81,7 @@ import org.apache.polaris.service.catalog.io.StorageAccessConfigProvider;
import org.apache.polaris.service.config.ReservedProperties;
import org.apache.polaris.service.context.catalog.PolarisPrincipalHolder;
import org.apache.polaris.service.events.PolarisEventMetadataFactory;
-import org.apache.polaris.service.events.listeners.NoOpPolarisEventListener;
+import org.apache.polaris.service.events.listeners.InMemoryEventCollector;
import org.apache.polaris.service.storage.PolarisStorageIntegrationProviderImpl;
import org.apache.polaris.service.task.TaskExecutor;
import org.apache.polaris.service.types.ApplicablePolicy;
@@ -251,7 +251,7 @@ public abstract class AbstractPolicyCatalogTest {
taskExecutor,
storageAccessConfigProvider,
fileIOFactory,
- new NoOpPolarisEventListener(),
+ new InMemoryEventCollector(),
eventMetadataFactory);
this.icebergCatalog.initialize(
CATALOG_NAME,
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/events/PolarisEventListenerOldConfigurationTest.java b/runtime/service/src/test/java/org/apache/polaris/service/events/PolarisEventListenerOldConfigurationTest.java
new file mode 100644
index 0000000000..311346766f
--- /dev/null
+++ b/runtime/service/src/test/java/org/apache/polaris/service/events/PolarisEventListenerOldConfigurationTest.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.polaris.service.events;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import com.google.common.collect.ImmutableMap;
+import io.quarkus.test.junit.QuarkusTest;
+import io.quarkus.test.junit.QuarkusTestProfile;
+import io.quarkus.test.junit.TestProfile;
+import jakarta.inject.Inject;
+import java.util.Map;
+import java.util.Set;
+import org.junit.jupiter.api.Test;
+
+@QuarkusTest
+@TestProfile(PolarisEventListenerOldConfigurationTest.OldConfigEventListenerProfile.class)
+public class PolarisEventListenerOldConfigurationTest {
+
+ public static class OldConfigEventListenerProfile implements QuarkusTestProfile {
+ @Override
+ public Map<String, String> getConfigOverrides() {
+ return ImmutableMap.<String, String>builder()
+ .put("polaris.event-listener.type", "test")
+ .build();
+ }
+ }
+
+ @Inject PolarisEventListenerConfiguration polarisEventListenerConfiguration;
+
+ @Test
+ public void testOldListenerConfig() {
+ assertTrue(polarisEventListenerConfiguration.type().isPresent());
+ assertTrue(polarisEventListenerConfiguration.types().isPresent());
+ assertEquals("test", polarisEventListenerConfiguration.type().get());
+ assertEquals(Set.of("test"), polarisEventListenerConfiguration.types().get());
+ }
+}
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/events/PolarisEventListenersMultipleListenersTest.java b/runtime/service/src/test/java/org/apache/polaris/service/events/PolarisEventListenersMultipleListenersTest.java
new file mode 100644
index 0000000000..615bdfa5fa
--- /dev/null
+++ b/runtime/service/src/test/java/org/apache/polaris/service/events/PolarisEventListenersMultipleListenersTest.java
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.polaris.service.events;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import com.google.common.collect.ImmutableMap;
+import io.quarkus.test.junit.QuarkusTest;
+import io.quarkus.test.junit.QuarkusTestProfile;
+import io.quarkus.test.junit.TestProfile;
+import io.smallrye.common.annotation.Identifier;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import org.apache.polaris.service.events.listeners.PolarisEventListener;
+import org.junit.jupiter.api.Test;
+
+@QuarkusTest
+@TestProfile(PolarisEventListenersMultipleListenersTest.TwoEventListenersProfile.class)
+public class PolarisEventListenersMultipleListenersTest {
+ static CountDownLatch latch = new CountDownLatch(2);
+
+ @ApplicationScoped
+ @Identifier("test-event-listener-1")
+ public static class EventListener1 implements PolarisEventListener {
+ boolean called = false;
+
+ @Override
+ public void onEvent(PolarisEvent event) {
+ if (called) {
+ throw new IllegalStateException("EventListener1 should only be called once");
+ }
+ called = true;
+ latch.countDown();
+ }
+ }
+
+ @ApplicationScoped
+ @Identifier("test-event-listener-2")
+ public static class EventListener2 implements PolarisEventListener {
+ boolean called = false;
+
+ @Override
+ public void onEvent(PolarisEvent event) {
+ if (called) {
+ throw new IllegalStateException("EventListener1 should only be called once");
+ }
+ called = true;
+ latch.countDown();
+ }
+ }
+
+ public static class TwoEventListenersProfile implements QuarkusTestProfile {
+ @Override
+ public Map<String, String> getConfigOverrides() {
+ return ImmutableMap.<String, String>builder()
+ .put("polaris.event-listener.types", "test-event-listener-1,test-event-listener-2")
+ .build();
+ }
+ }
+
+ @Inject PolarisEventDispatcher eventDispatcher;
+
+ @Test
+ public void testAllListenersGetNotified() throws InterruptedException {
+ eventDispatcher.dispatch(new PolarisEvent(PolarisEventType.AFTER_SEND_NOTIFICATION, null));
+ assertTrue(latch.await(5, TimeUnit.SECONDS));
+ }
+}
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/ratelimiter/RateLimiterFilterTest.java b/runtime/service/src/test/java/org/apache/polaris/service/ratelimiter/RateLimiterFilterTest.java
index e258653b25..9f4204a8fd 100644
--- a/runtime/service/src/test/java/org/apache/polaris/service/ratelimiter/RateLimiterFilterTest.java
+++ b/runtime/service/src/test/java/org/apache/polaris/service/ratelimiter/RateLimiterFilterTest.java
@@ -26,6 +26,7 @@ import io.micrometer.core.instrument.MeterRegistry;
import io.quarkus.test.junit.QuarkusTest;
import io.quarkus.test.junit.QuarkusTestProfile;
import io.quarkus.test.junit.TestProfile;
+import io.smallrye.common.annotation.Identifier;
import jakarta.inject.Inject;
import jakarta.ws.rs.core.Response.Status;
import java.util.Map;
@@ -78,14 +79,19 @@ public class RateLimiterFilterTest {
.put("polaris.realm-context.type", "test")
.put("polaris.authentication.token-broker.type", "symmetric-key")
.put("polaris.authentication.token-broker.symmetric-key.secret", "secret")
- .put("polaris.event-listener.type", "test")
+ .put("polaris.event-listener.types", "test")
.build();
}
}
@Inject PolarisIntegrationTestHelper helper;
@Inject MeterRegistry meterRegistry;
- @Inject PolarisEventListener polarisEventListener;
+
+ @Inject
+ @Identifier("test")
+ PolarisEventListener polarisEventListener;
+
+ private TestPolarisEventListener testPolarisEventListener;
private TestEnvironment testEnv;
private PolarisIntegrationTestFixture fixture;
@@ -108,6 +114,9 @@ public class RateLimiterFilterTest {
public void resetMeterRegistry() {
MockRateLimiter.allowProceed = true;
meterRegistry.clear();
+
+ testPolarisEventListener = (TestPolarisEventListener) polarisEventListener;
+ testPolarisEventListener.clear();
}
@Test
@@ -142,8 +151,7 @@ public class RateLimiterFilterTest {
}
PolarisEvent event =
- ((TestPolarisEventListener) polarisEventListener)
- .getLatest(PolarisEventType.BEFORE_LIMIT_REQUEST_RATE);
+ testPolarisEventListener.getLatest(PolarisEventType.BEFORE_LIMIT_REQUEST_RATE);
assertThat(event.attributes().getRequired(EventAttributes.HTTP_METHOD)).isEqualTo("GET");
// Examples of expected metrics:
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/task/TaskExecutorImplTest.java b/runtime/service/src/test/java/org/apache/polaris/service/task/TaskExecutorImplTest.java
index d2ddd29dcb..63b4a46205 100644
--- a/runtime/service/src/test/java/org/apache/polaris/service/task/TaskExecutorImplTest.java
+++ b/runtime/service/src/test/java/org/apache/polaris/service/task/TaskExecutorImplTest.java
@@ -30,7 +30,7 @@ import org.apache.polaris.service.events.EventAttributes;
import org.apache.polaris.service.events.PolarisEvent;
import org.apache.polaris.service.events.PolarisEventMetadata;
import org.apache.polaris.service.events.PolarisEventType;
-import org.apache.polaris.service.events.listeners.TestPolarisEventListener;
+import org.apache.polaris.service.events.listeners.InMemoryEventCollector;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -43,8 +43,8 @@ public class TaskExecutorImplTest {
TestServices testServices = TestServices.builder().realmContext(realmContext).build();
- TestPolarisEventListener testPolarisEventListener =
- (TestPolarisEventListener) testServices.polarisEventListener();
+ InMemoryEventCollector testPolarisEventDispatcher =
+ (InMemoryEventCollector) testServices.polarisEventDispatcher();
PolarisMetaStoreManager metaStoreManager = testServices.metaStoreManager();
@@ -71,7 +71,7 @@ public class TaskExecutorImplTest {
new TaskFileIOSupplier(
testServices.fileIOFactory(), testServices.storageAccessConfigProvider()),
new RealmContextHolder(),
- testServices.polarisEventListener(),
+ testServices.polarisEventDispatcher(),
testServices.eventMetadataFactory(),
null,
new PolarisPrincipalHolder(),
@@ -87,7 +87,7 @@ public class TaskExecutorImplTest {
@Override
public boolean handleTask(TaskEntity task, CallContext callContext) {
PolarisEvent beforeTaskAttemptedEvent =
- testPolarisEventListener.getLatest(PolarisEventType.BEFORE_ATTEMPT_TASK);
+ testPolarisEventDispatcher.getLatest(PolarisEventType.BEFORE_ATTEMPT_TASK);
Assertions.assertEquals(
taskEntity.getId(),
beforeTaskAttemptedEvent.attributes().getRequired(EventAttributes.TASK_ENTITY_ID));
@@ -105,7 +105,7 @@ public class TaskExecutorImplTest {
attempt);
PolarisEvent afterAttemptTaskEvent =
- testPolarisEventListener.getLatest(PolarisEventType.AFTER_ATTEMPT_TASK);
+ testPolarisEventDispatcher.getLatest(PolarisEventType.AFTER_ATTEMPT_TASK);
Assertions.assertEquals(
taskEntity.getId(),
afterAttemptTaskEvent.attributes().getRequired(EventAttributes.TASK_ENTITY_ID));
diff --git a/runtime/service/src/testFixtures/java/org/apache/polaris/service/TestServices.java b/runtime/service/src/testFixtures/java/org/apache/polaris/service/TestServices.java
index e6310a09e7..de597b8685 100644
--- a/runtime/service/src/testFixtures/java/org/apache/polaris/service/TestServices.java
+++ b/runtime/service/src/testFixtures/java/org/apache/polaris/service/TestServices.java
@@ -85,10 +85,10 @@ import org.apache.polaris.service.context.catalog.PolarisCallContextCatalogFacto
import org.apache.polaris.service.credentials.DefaultPolarisCredentialManager;
import org.apache.polaris.service.credentials.connection.SigV4ConnectionCredentialVendor;
import org.apache.polaris.service.events.EventAttributeMap;
+import org.apache.polaris.service.events.PolarisEventDispatcher;
import org.apache.polaris.service.events.PolarisEventMetadata;
import org.apache.polaris.service.events.PolarisEventMetadataFactory;
-import org.apache.polaris.service.events.listeners.PolarisEventListener;
-import org.apache.polaris.service.events.listeners.TestPolarisEventListener;
+import org.apache.polaris.service.events.listeners.InMemoryEventCollector;
import org.apache.polaris.service.identity.provider.DefaultServiceIdentityProvider;
import org.apache.polaris.service.persistence.InMemoryPolarisMetaStoreManagerFactory;
import org.apache.polaris.service.reporting.DefaultMetricsReporter;
@@ -120,7 +120,7 @@ public record TestServices(
PolarisMetaStoreManager metaStoreManager,
FileIOFactory fileIOFactory,
TaskExecutor taskExecutor,
- PolarisEventListener polarisEventListener,
+ PolarisEventDispatcher polarisEventDispatcher,
PolarisEventMetadataFactory eventMetadataFactory,
StorageAccessConfigProvider storageAccessConfigProvider) {
@@ -298,7 +298,7 @@ public record TestServices(
TaskExecutor taskExecutor = Mockito.mock(TaskExecutor.class);
- PolarisEventListener polarisEventListener = new TestPolarisEventListener();
+ PolarisEventDispatcher polarisEventDispatcher = new InMemoryEventCollector();
CallContextCatalogFactory callContextFactory =
new PolarisCallContextCatalogFactory(
diagnostics,
@@ -306,7 +306,7 @@ public record TestServices(
taskExecutor,
storageAccessConfigProvider,
fileIOFactory,
- polarisEventListener,
+ polarisEventDispatcher,
eventMetadataFactory,
metaStoreManager,
callContext,
@@ -362,13 +362,13 @@ public record TestServices(
finalRestCatalogService =
new IcebergRestCatalogEventServiceDelegator(
catalogService,
- polarisEventListener,
+ polarisEventDispatcher,
eventMetadataFactory,
new DefaultCatalogPrefixParser(),
eventAttributeMap);
finalRestConfigurationService =
new IcebergRestConfigurationEventServiceDelegator(
- catalogService, polarisEventListener, eventMetadataFactory);
+ catalogService, polarisEventDispatcher, eventMetadataFactory);
}
IcebergRestCatalogApi restApi = new IcebergRestCatalogApi(finalRestCatalogService);
@@ -409,7 +409,7 @@ public record TestServices(
metaStoreManager,
fileIOFactory,
taskExecutor,
- polarisEventListener,
+ polarisEventDispatcher,
eventMetadataFactory,
storageAccessConfigProvider);
}
diff --git a/runtime/service/src/main/java/org/apache/polaris/service/events/listeners/NoOpPolarisEventListener.java b/runtime/service/src/testFixtures/java/org/apache/polaris/service/events/listeners/InMemoryEventCollector.java
similarity index 51%
rename from runtime/service/src/main/java/org/apache/polaris/service/events/listeners/NoOpPolarisEventListener.java
rename to runtime/service/src/testFixtures/java/org/apache/polaris/service/events/listeners/InMemoryEventCollector.java
index bfef6e96ec..91eaddb29b 100644
--- a/runtime/service/src/main/java/org/apache/polaris/service/events/listeners/NoOpPolarisEventListener.java
+++ b/runtime/service/src/testFixtures/java/org/apache/polaris/service/events/listeners/InMemoryEventCollector.java
@@ -16,12 +16,32 @@
* specific language governing permissions and limitations
* under the License.
*/
+
package org.apache.polaris.service.events.listeners;
-import io.smallrye.common.annotation.Identifier;
-import jakarta.enterprise.context.ApplicationScoped;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.polaris.service.events.PolarisEvent;
+import org.apache.polaris.service.events.PolarisEventDispatcher;
+import org.apache.polaris.service.events.PolarisEventType;
+
+public class InMemoryEventCollector implements PolarisEventDispatcher {
+ private final Map<PolarisEventType, PolarisEvent> latestEvents = new ConcurrentHashMap<>();
+
+ public void clear() {
+ latestEvents.clear();
+ }
+
+ public PolarisEvent getLatest(PolarisEventType type) {
+ var latest = latestEvents.get(type);
+ if (latest == null) {
+ throw new IllegalStateException("No event of type " + type + " recorded");
+ }
+ return latest;
+ }
-/** Event listener that does nothing. */
-@ApplicationScoped
-@Identifier("no-op")
-public class NoOpPolarisEventListener implements PolarisEventListener {}
+ @Override
+ public void dispatch(PolarisEvent event) {
+ latestEvents.put(event.type(), event);
+ }
+}
diff --git a/site/content/in-dev/unreleased/configuration/config-sections/smallrye-polaris_event_listener.md b/site/content/in-dev/unreleased/configuration/config-sections/smallrye-polaris_event_listener.md
index ef266451e8..ec83859542 100644
--- a/site/content/in-dev/unreleased/configuration/config-sections/smallrye-polaris_event_listener.md
+++ b/site/content/in-dev/unreleased/configuration/config-sections/smallrye-polaris_event_listener.md
@@ -25,4 +25,5 @@ build:
| Property | Default Value | Type | Description |
|----------|---------------|------|-------------|
-| `polaris.event-listener.type` | | `string` | The type of the event listener to use. Must be a registered (`PolarisEventListener`) identifier. |
+| `polaris.event-listener.type` | | `string` | The type of the event listener to use. Must be a registered (`PolarisEventListener`) identifier. <br><br>_Deprecated_ since 1.5.0, use 'polaris.event-listener.types' instead |
+| `polaris.event-listener.types` | | `list of string` | Comma separated list of event listers, each item must be a registered (`PolarisEventListener`) identifier. |