You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@polaris.apache.org by "sririshindra (via GitHub)" <gi...@apache.org> on 2026/04/16 20:58:53 UTC
[PR] Refactor persistence event listener to strategy-based table handlers and add dedicated table handler classes [polaris]
sririshindra opened a new pull request, #4225:
URL: https://github.com/apache/polaris/pull/4225
Replace switch-based dispatch with handler registry for persistence events.
Add dedicated catalog/table handler classes covering all table events (600-617).
Add exhaustive listener tests for routing and additional property behaviors.
<!--
๐ Describe what changes you're proposing, especially breaking or user-facing changes.
๐ See https://github.com/apache/polaris/blob/main/CONTRIBUTING.md for more.
-->
## Checklist
- [x] ๐ก๏ธ Don't disclose security issues! (contact security@apache.org)
- [x] ๐ Clearly explained why the changes are needed, or linked related issues: Fixes #
- [x] ๐งช Added/updated tests with good coverage, or manually tested (and explained how)
- [x] ๐ก Added comments for complex logic
- [ ] ๐งพ Updated `CHANGELOG.md` (if needed)
- [ ] ๐ Updated documentation in `site/content/in-dev/unreleased` (if needed)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor persistence event listener to strategy-based table handlers and add dedicated table handler classes [polaris]
Posted by "nandorKollar (via GitHub)" <gi...@apache.org>.
nandorKollar commented on PR #4225:
URL: https://github.com/apache/polaris/pull/4225#issuecomment-4266350649
I think that instead of replacing the switch-case to event handlers, we should think about something which scales better. With event handlers, we still possibly need more than 150 event handler to transform each service event type. I think the main reason, why we need separate handlers for each even type is that the content of the EventAttributeMap is different for each type. What if we say this: we serialize every possible AttributeKey to Json, which is present in the EventAttributeMap, and set the additionalProperties to the serialised value on the PolarisEvent entity. Something like this:
```
private static final ObjectMapper MAPPER = JsonMapper.builder().build();
private String convertAttributesToJson(PolarisEvent event) {
Map<String, Object> values = new HashMap<>();
for (AttributeKey<?> a : EventAttributes.ALL) {
addAttribute(values, event, a);
}
try {
return MAPPER.writeValueAsString(values);
} catch (JsonProcessingException e) {
throw new RuntimeException(e);
}
}
private void addAttribute(Map<String, Object> values, PolarisEvent event, AttributeKey<?> attributeKey) {
var attributeValue = event.attributes().get(attributeKey);
attributeValue.ifPresent(value -> values.put(attributeKey.name(), value));
}
```
We should find a way to customise the ObjectMapper, so that we can whitelist properties for each object to allow for serialisation, as the attribute values can contain sensitive data, which we don't want to save in the database.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor persistence event listener to strategy-based table handlers and add dedicated table handler classes [polaris]
Posted by "adutra (via GitHub)" <gi...@apache.org>.
adutra commented on PR #4225:
URL: https://github.com/apache/polaris/pull/4225#issuecomment-4269742934
I agree with @nandorKollar, the handlers approach is just transforming 150+ switch cases into 150+ handlers.
A common mechanism that would be able to persist all events sounds definitely better to me.
Also, I think we need to get #4064 merged first, as it brings substantial improvements.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor persistence event listener to a universal registry pattern with strict allowlist serialization [polaris]
Posted by "youngrae81 (via GitHub)" <gi...@apache.org>.
youngrae81 commented on code in PR #4225:
URL: https://github.com/apache/polaris/pull/4225#discussion_r3128597910
##########
runtime/service/src/main/java/org/apache/polaris/service/events/listeners/PersistenceAttributeSecurityFilter.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.listeners;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Set;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableMetadataParser;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.rest.responses.LoadTableResponse;
+import org.apache.polaris.service.events.AttributeKey;
+import org.apache.polaris.service.events.EventAttributeMap;
+import org.apache.polaris.service.events.EventAttributes;
+
+final class PersistenceAttributeSecurityFilter {
+
+ private static final Set<String> ALLOWED_TELEMETRY_CONTEXT_KEYS =
+ Set.of("otel.trace_id", "otel.span_id", "otel.trace_flags", "otel.sampled");
+
+ Map<String, String> filterAllowlistedAttributes(EventAttributeMap attributes) {
+ Map<String, String> safeAttributes = new LinkedHashMap<>();
+ attributes.forEach((key, value) -> addIfAllowlisted(safeAttributes, key, value));
+ return safeAttributes;
+ }
+
+ Map<String, String> filterAllowlistedTelemetryContext(Map<String, String> telemetryContext) {
+ Map<String, String> safeContext = new LinkedHashMap<>();
+ telemetryContext.forEach(
+ (key, value) -> {
+ if (ALLOWED_TELEMETRY_CONTEXT_KEYS.contains(key)) {
+ safeContext.put(key, value);
+ }
+ });
+ return safeContext;
+ }
+
+ @SuppressWarnings("unchecked")
+ private static void addIfAllowlisted(
+ Map<String, String> safeAttributes, AttributeKey<?> key, Object value) {
+ if (EventAttributes.CATALOG_NAME.equals(key)
+ || EventAttributes.TABLE_NAME.equals(key)
+ || EventAttributes.VIEW_NAME.equals(key)
+ || EventAttributes.NAMESPACE_FQN.equals(key)
+ || EventAttributes.PARENT_NAMESPACE_FQN.equals(key)) {
+ safeAttributes.put(key.name(), (String) value);
+ return;
+ }
+
+ if (EventAttributes.NAMESPACE.equals(key)) {
+ safeAttributes.put(key.name(), ((Namespace) value).toString());
+ return;
+ }
+
+ if (EventAttributes.TABLE_IDENTIFIER.equals(key)
+ || EventAttributes.VIEW_IDENTIFIER.equals(key)) {
+ safeAttributes.put(key.name(), ((TableIdentifier) value).toString());
+ return;
+ }
+
+ // TODO: Filter sensitive events and credential-bearing table properties before serializing
+ // table metadata. This and other sensitive event filtering cases will be addressed separately.
+ if (EventAttributes.TABLE_METADATA.equals(key)) {
+ addTableMetadata(safeAttributes, (TableMetadata) value);
+ return;
+ }
+
+ if (EventAttributes.LOAD_TABLE_RESPONSE.equals(key)) {
+ addTableMetadata(safeAttributes, ((LoadTableResponse) value).tableMetadata());
+ return;
+ }
Review Comment:
RENAME_TABLE_REQUEST is used in DefaultPersistenceEventHandler.resolveTableLikeIdentifier() to extract the source table, but it's not in the allowlist here.
This means for AFTER_RENAME_TABLE events, the destination table identifier won't appear in additionalProperties โ only the source is captured via resourceIdentifier.
Should the destination be allowlisted too?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor persistence event listener to a universal registry pattern with strict allowlist serialization [polaris]
Posted by "sririshindra (via GitHub)" <gi...@apache.org>.
sririshindra commented on code in PR #4225:
URL: https://github.com/apache/polaris/pull/4225#discussion_r3231032215
##########
runtime/service/src/main/java/org/apache/polaris/service/events/listeners/PersistenceAttributeSecurityFilter.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.listeners;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Set;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableMetadataParser;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.rest.responses.LoadTableResponse;
+import org.apache.polaris.service.events.AttributeKey;
+import org.apache.polaris.service.events.EventAttributeMap;
+import org.apache.polaris.service.events.EventAttributes;
+
+final class PersistenceAttributeSecurityFilter {
+
+ private static final Set<String> ALLOWED_TELEMETRY_CONTEXT_KEYS =
+ Set.of("otel.trace_id", "otel.span_id", "otel.trace_flags", "otel.sampled");
+
+ Map<String, String> filterAllowlistedAttributes(EventAttributeMap attributes) {
+ Map<String, String> safeAttributes = new LinkedHashMap<>();
+ attributes.forEach((key, value) -> addIfAllowlisted(safeAttributes, key, value));
+ return safeAttributes;
+ }
+
+ Map<String, String> filterAllowlistedTelemetryContext(Map<String, String> telemetryContext) {
+ Map<String, String> safeContext = new LinkedHashMap<>();
+ telemetryContext.forEach(
+ (key, value) -> {
+ if (ALLOWED_TELEMETRY_CONTEXT_KEYS.contains(key)) {
+ safeContext.put(key, value);
+ }
+ });
+ return safeContext;
+ }
+
+ @SuppressWarnings("unchecked")
+ private static void addIfAllowlisted(
+ Map<String, String> safeAttributes, AttributeKey<?> key, Object value) {
+ if (EventAttributes.CATALOG_NAME.equals(key)
+ || EventAttributes.TABLE_NAME.equals(key)
+ || EventAttributes.VIEW_NAME.equals(key)
+ || EventAttributes.NAMESPACE_FQN.equals(key)
+ || EventAttributes.PARENT_NAMESPACE_FQN.equals(key)) {
+ safeAttributes.put(key.name(), (String) value);
+ return;
+ }
+
+ if (EventAttributes.NAMESPACE.equals(key)) {
+ safeAttributes.put(key.name(), ((Namespace) value).toString());
+ return;
+ }
+
+ if (EventAttributes.TABLE_IDENTIFIER.equals(key)
+ || EventAttributes.VIEW_IDENTIFIER.equals(key)) {
+ safeAttributes.put(key.name(), ((TableIdentifier) value).toString());
+ return;
+ }
+
+ // TODO: Filter sensitive events and credential-bearing table properties before serializing
+ // table metadata. This and other sensitive event filtering cases will be addressed separately.
+ if (EventAttributes.TABLE_METADATA.equals(key)) {
+ addTableMetadata(safeAttributes, (TableMetadata) value);
+ return;
+ }
+
+ if (EventAttributes.LOAD_TABLE_RESPONSE.equals(key)) {
+ addTableMetadata(safeAttributes, ((LoadTableResponse) value).tableMetadata());
+ return;
+ }
Review Comment:
Good catch. RENAME_TABLE_REQUEST is now in the allowlist and the DefaultEventPayloadPruner serializes it as two explicit keys: rename_source and rename_destination. Additionally, resourceIdentifier captures source (for BEFORE_ events) or destination (for AFTER_ events). Both values are fully captured.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor persistence event listener to a universal registry pattern with strict allowlist serialization [polaris]
Posted by "sririshindra (via GitHub)" <gi...@apache.org>.
sririshindra commented on code in PR #4225:
URL: https://github.com/apache/polaris/pull/4225#discussion_r3231021194
##########
runtime/service/src/main/java/org/apache/polaris/service/events/listeners/DefaultPersistenceEventHandler.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * 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.listeners;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Objects;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.rest.requests.RenameTableRequest;
+import org.apache.polaris.core.admin.model.Catalog;
+import org.apache.polaris.core.admin.model.Principal;
+import org.apache.polaris.core.auth.PolarisPrincipal;
+import org.apache.polaris.core.entity.PolarisEvent.ResourceType;
+import org.apache.polaris.service.events.EventAttributeMap;
+import org.apache.polaris.service.events.EventAttributes;
+import org.apache.polaris.service.events.PolarisEvent;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+final class DefaultPersistenceEventHandler implements PersistenceEventHandler {
+
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(DefaultPersistenceEventHandler.class);
+ private static final String SERIALIZATION_ERROR_PAYLOAD = "{\"error\":\"Serialization failed\"}";
+
+ static final String ALL_CATALOGS_SCOPE = "_all_catalogs";
+ static final String ROOT_NAMESPACE_SCOPE = "_root_namespace";
+ static final String TABLE_SCOPE = "_table_scope";
+ static final String VIEW_SCOPE = "_view_scope";
+
+ private final PersistenceAttributeSecurityFilter securityFilter;
+ private final ObjectMapper objectMapper;
+
+ DefaultPersistenceEventHandler() {
+ this(new PersistenceAttributeSecurityFilter(), new ObjectMapper());
Review Comment:
Addressed. The static ObjectMapper is removed. The EventPayloadPruner and EventAttributeFilter are now CDI beans injected into PolarisPersistenceEventListener. The entity's own setAdditionalProperties(Map) handles JSON serialization using its internal mapper, so the listener no longer needs a direct ObjectMapper reference.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Overhaul event persistence with universal routing and global secure-by-default sanitization [polaris]
Posted by "nandorKollar (via GitHub)" <gi...@apache.org>.
nandorKollar commented on code in PR #4225:
URL: https://github.com/apache/polaris/pull/4225#discussion_r3367047502
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -80,8 +82,13 @@ public void onStartup(@Observes StartupEvent event) {
private void deliverEvent(
PolarisEvent event, String listenerName, PolarisEventListener listener) {
LOGGER.debug("Delivering {} event to listener '{}' ({})", event.type(), listenerName, listener);
+ // TODO: Add an opt-in mechanism (e.g., a marker interface or annotation) for specialized
Review Comment:
> I think we are not plan to apply different sanitizer to different event listener, right?
I think for now a single, global sanitizes should be sufficient.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor persistence event listener to strategy-based table handlers and add dedicated table handler classes [polaris]
Posted by "youngrae81 (via GitHub)" <gi...@apache.org>.
youngrae81 commented on PR #4225:
URL: https://github.com/apache/polaris/pull/4225#issuecomment-4284938894
I'd like to share some practical experience relevant to the design discussion here.
I implemented 13 namespace/table handlers locally (#4227) following the current switch-case pattern.
The result was ~300 lines of nearly identical boilerplate โ each handler extracts catalogName, builds a PolarisEvent with ResourceType, and adds OTel context. The only variation is which attributes go into additionalProperties.
This firsthand experience reinforces the concerns raised by @nandorKollar and @adutra. I support the hybrid approach suggested by @obelix74
1. A generic fallback handler that serializes EventAttributeMap values as JSON โ covers 90%+ of events with zero per-type code
2. Specialized handlers only where structured fields are needed (e.g., table-uuid from TableMetadata, source/destination for RENAME)
Happy to help once the design direction is finalized.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor persistence event listener to strategy-based table handlers and add dedicated table handler classes [polaris]
Posted by "sririshindra (via GitHub)" <gi...@apache.org>.
sririshindra commented on PR #4225:
URL: https://github.com/apache/polaris/pull/4225#issuecomment-4269863725
> I agree with @nandorKollar, the handlers approach is just transforming 150+ switch cases into 150+ handlers.
>
> A common mechanism that would be able to persist all events sounds definitely better to me.
>
> Also, I think we need to get #4064 merged first, as it brings substantial improvements.
Thanks @nandorKollar and @adutra . I was half way through implementing a common mechanism already but I will wait till #4064 is merged and rebase first. Thanks for pointing it out.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor persistence event listener to a universal registry pattern with strict allowlist serialization [polaris]
Posted by "sririshindra (via GitHub)" <gi...@apache.org>.
sririshindra commented on code in PR #4225:
URL: https://github.com/apache/polaris/pull/4225#discussion_r3231017166
##########
runtime/service/src/main/java/org/apache/polaris/service/events/listeners/PersistenceEventHandler.java:
##########
@@ -0,0 +1,27 @@
+/*
+ * 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.listeners;
+
+import org.apache.polaris.service.events.PolarisEvent;
+
+interface PersistenceEventHandler {
Review Comment:
Thanks @adutra โ I agree this should ultimately be a global concern. In this revision, I've extracted the filtering and pruning logic into two public interfaces at the events package level (not within listeners):
- EventAttributeFilter โ decides which AttributeKeys are safe to pass downstream
- EventPayloadPruner โ reduces large attribute values (especially TableMetadata) to bounded summaries
Both are CDI beans with no coupling to persistence. They can be promoted to PolarisEventListeners.deliverEvent() as a global pre-delivery sanitizer once we reach community consensus on that approach. I will send an email to the mailing list for this.
For now they're functional within the persistence listener but architecturally ready to move.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Overhaul event persistence with universal routing and global secure-by-default sanitization [polaris]
Posted by "nandorKollar (via GitHub)" <gi...@apache.org>.
nandorKollar commented on code in PR #4225:
URL: https://github.com/apache/polaris/pull/4225#discussion_r3298889781
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -80,8 +82,13 @@ public void onStartup(@Observes StartupEvent event) {
private void deliverEvent(
PolarisEvent event, String listenerName, PolarisEventListener listener) {
LOGGER.debug("Delivering {} event to listener '{}' ({})", event.type(), listenerName, listener);
+ // TODO: Add an opt-in mechanism (e.g., a marker interface or annotation) for specialized
Review Comment:
I think the question is, do should we allow this case, or not? With the current approach, we don't let any listener receive unsanitized events. Why do we submit the event with sensitive data all, when eventually no listener can access the unsanitized event at all?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor persistence event listener to a universal registry pattern with strict allowlist serialization [polaris]
Posted by "nandorKollar (via GitHub)" <gi...@apache.org>.
nandorKollar commented on code in PR #4225:
URL: https://github.com/apache/polaris/pull/4225#discussion_r3179225448
##########
runtime/service/src/main/java/org/apache/polaris/service/events/listeners/DefaultPersistenceEventHandler.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * 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.listeners;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Objects;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.rest.requests.RenameTableRequest;
+import org.apache.polaris.core.admin.model.Catalog;
+import org.apache.polaris.core.admin.model.Principal;
+import org.apache.polaris.core.auth.PolarisPrincipal;
+import org.apache.polaris.core.entity.PolarisEvent.ResourceType;
+import org.apache.polaris.service.events.EventAttributeMap;
+import org.apache.polaris.service.events.EventAttributes;
+import org.apache.polaris.service.events.PolarisEvent;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+final class DefaultPersistenceEventHandler implements PersistenceEventHandler {
+
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(DefaultPersistenceEventHandler.class);
+ private static final String SERIALIZATION_ERROR_PAYLOAD = "{\"error\":\"Serialization failed\"}";
+
+ static final String ALL_CATALOGS_SCOPE = "_all_catalogs";
+ static final String ROOT_NAMESPACE_SCOPE = "_root_namespace";
+ static final String TABLE_SCOPE = "_table_scope";
+ static final String VIEW_SCOPE = "_view_scope";
+
+ private final PersistenceAttributeSecurityFilter securityFilter;
+ private final ObjectMapper objectMapper;
+
+ DefaultPersistenceEventHandler() {
+ this(new PersistenceAttributeSecurityFilter(), new ObjectMapper());
+ }
+
+ DefaultPersistenceEventHandler(
+ PersistenceAttributeSecurityFilter securityFilter, ObjectMapper objectMapper) {
+ this.securityFilter = Objects.requireNonNull(securityFilter, "securityFilter");
+ this.objectMapper = Objects.requireNonNull(objectMapper, "objectMapper");
+ }
+
+ @Override
+ public org.apache.polaris.core.entity.PolarisEvent toPersistenceEvent(PolarisEvent event) {
+ EventAttributeMap attributes = event.attributes();
+ String catalogName = resolveCatalogName(attributes);
+ ResourceType resourceType = resolveResourceType(event, attributes);
+
+ org.apache.polaris.core.entity.PolarisEvent persistenceEvent =
+ new org.apache.polaris.core.entity.PolarisEvent(
+ catalogName,
+ event.metadata().eventId().toString(),
+ event.metadata().requestId().orElse(null),
+ event.type().name(),
+ event.metadata().timestamp().toEpochMilli(),
+ event.metadata().user().map(PolarisPrincipal::getName).orElse(null),
+ resourceType,
+ resolveResourceIdentifier(resourceType, attributes, catalogName));
+
+ Map<String, String> additionalProperties =
+ new LinkedHashMap<>(securityFilter.filterAllowlistedAttributes(attributes));
+ additionalProperties.putAll(
+ securityFilter.filterAllowlistedTelemetryContext(event.metadata().openTelemetryContext()));
+ if (!additionalProperties.isEmpty()) {
+ persistenceEvent.setAdditionalProperties(
+ serializeAdditionalProperties(event, additionalProperties));
+ }
+
+ return persistenceEvent;
+ }
+
+ private String serializeAdditionalProperties(
+ PolarisEvent event, Map<String, String> additionalProperties) {
+ try {
+ return objectMapper.writeValueAsString(additionalProperties);
+ } catch (JsonProcessingException e) {
+ LOGGER.warn(
+ "Failed to serialize additional properties for eventType={} eventId={}; persisting fallback payload",
+ event.type(),
+ event.metadata().eventId());
+ LOGGER.debug(
+ "Serialization failure details for eventType={} eventId={}",
+ event.type(),
+ event.metadata().eventId(),
+ e);
+ return SERIALIZATION_ERROR_PAYLOAD;
+ }
+ }
+
+ private static String resolveCatalogName(EventAttributeMap attributes) {
+ return attributes
+ .get(EventAttributes.CATALOG_NAME)
+ .or(() -> attributes.get(EventAttributes.CATALOG).map(Catalog::getName))
+ .orElse(ALL_CATALOGS_SCOPE);
+ }
+
+ private static ResourceType resolveResourceType(
+ PolarisEvent event, EventAttributeMap attributes) {
+ return switch (event.type().category()) {
+ case CATALOG -> ResourceType.CATALOG;
+ case NAMESPACE -> ResourceType.NAMESPACE;
+ case TABLE -> ResourceType.TABLE;
+ case VIEW -> ResourceType.VIEW;
+ // Persistence resource type schema supports CATALOG/NAMESPACE/TABLE/VIEW only.
Review Comment:
I agree, event persistence should not selectively filter out event types or categories. This is already configurable, if someone would like to persist only a subset of event types.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Overhaul event persistence with universal routing and global secure-by-default sanitization [polaris]
Posted by "sririshindra (via GitHub)" <gi...@apache.org>.
sririshindra commented on code in PR #4225:
URL: https://github.com/apache/polaris/pull/4225#discussion_r3382995720
##########
runtime/service/src/main/java/org/apache/polaris/service/events/listeners/PolarisPersistenceEventListener.java:
##########
@@ -19,38 +19,37 @@
package org.apache.polaris.service.events.listeners;
-import com.google.common.collect.ImmutableMap;
+import jakarta.inject.Inject;
+import java.util.LinkedHashMap;
import java.util.Map;
-import org.apache.iceberg.TableMetadata;
-import org.apache.iceberg.TableMetadataParser;
+import java.util.Optional;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
-import org.apache.iceberg.rest.responses.LoadTableResponse;
-import org.apache.polaris.core.admin.model.Catalog;
+import org.apache.iceberg.rest.requests.CreateTableRequest;
+import org.apache.iceberg.rest.requests.RegisterTableRequest;
import org.apache.polaris.core.auth.PolarisPrincipal;
+import org.apache.polaris.service.events.EventAttributeMap;
import org.apache.polaris.service.events.EventAttributes;
+import org.apache.polaris.service.events.EventPayloadPruner;
import org.apache.polaris.service.events.PolarisEvent;
+import org.apache.polaris.service.events.PolarisEventType;
public abstract class PolarisPersistenceEventListener implements PolarisEventListener {
- @Override
- public void onEvent(PolarisEvent event) {
- switch (event.type()) {
- case AFTER_CREATE_TABLE -> handleAfterCreateTable(event);
- case AFTER_CREATE_CATALOG -> handleAfterCreateCatalog(event);
- default -> {
- // Other events not handled by this listener
- }
- }
+ @Inject EventPayloadPruner payloadPruner;
+
+ protected PolarisPersistenceEventListener() {}
+
+ PolarisPersistenceEventListener(EventPayloadPruner payloadPruner) {
+ this.payloadPruner = payloadPruner;
}
- private void handleAfterCreateTable(PolarisEvent event) {
- LoadTableResponse loadTableResponse =
- event.attributes().getRequired(EventAttributes.LOAD_TABLE_RESPONSE);
- TableMetadata tableMetadata = loadTableResponse.tableMetadata();
- String catalogName = event.attributes().getRequired(EventAttributes.CATALOG_NAME);
- Namespace namespace = event.attributes().getRequired(EventAttributes.NAMESPACE);
- String tableName = event.attributes().getRequired(EventAttributes.TABLE_NAME);
+ @Override
+ public void onEvent(PolarisEvent event) {
+ String catalogName = resolveCatalogName(event);
Review Comment:
Filed https://github.com/apache/polaris/issues/4674
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Overhaul event persistence with universal routing and global secure-by-default sanitization [polaris]
Posted by "sririshindra (via GitHub)" <gi...@apache.org>.
sririshindra commented on PR #4225:
URL: https://github.com/apache/polaris/pull/4225#issuecomment-4661869834
> Hi @sririshindra, thank you for the thorough work on this! I also appreciate the patience you've shown in addressing all the comments and reviewer feedback.
>
> I believe this PR introduces significant enhancements, most notably that the persistence listener can now handle all event types. Given that it has already received other approvals, I suggest we proceed with the merge, as it represents a clear improvement over our current state.
>
> That said, as mentioned in my comments, the design still feels slightly rigid and doesn't yet fully capitalize on the new `EventSanitizer` idea. My long-term goal is to reach a state where sanitizers and listeners form a fully configurable event delivery pipeline that doesn't require rebuilding Polaris. I think that CDI and the Vert.x event bus can deliver that.
>
> Thanks again for the hard work; we can address my design ideas in subsequent follow-up tasks.
Thank you @adutra for your thorough review. You mentioned we can merge this PR and address your design ideas in subsequent follow-ups. That sounds like a good plan to me.
Is there any change that you would like me to do in this PR itself and not a follow-up? If not, can you approve this PR so that it can get merged and I will address your comments in a follow-up? Does that work?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor persistence event listener to a universal registry pattern with strict allowlist serialization [polaris]
Posted by "nandorKollar (via GitHub)" <gi...@apache.org>.
nandorKollar commented on code in PR #4225:
URL: https://github.com/apache/polaris/pull/4225#discussion_r3154624183
##########
runtime/service/src/main/java/org/apache/polaris/service/events/listeners/PersistenceEventHandler.java:
##########
@@ -0,0 +1,27 @@
+/*
+ * 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.listeners;
+
+import org.apache.polaris.service.events.PolarisEvent;
+
+interface PersistenceEventHandler {
Review Comment:
@adutra do you happen to have a reference for this discussion?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Overhaul event persistence with universal routing and global secure-by-default sanitization [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on PR #4225:
URL: https://github.com/apache/polaris/pull/4225#issuecomment-4634700315
@flyrain : It would be nice to give @adutra some more time to have another look (based on previous involvement in reviewing this PR)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Overhaul event persistence with universal routing and global secure-by-default sanitization [polaris]
Posted by "flyrain (via GitHub)" <gi...@apache.org>.
flyrain commented on code in PR #4225:
URL: https://github.com/apache/polaris/pull/4225#discussion_r3358819148
##########
runtime/service/src/main/java/org/apache/polaris/service/events/EventAttributeFilter.java:
##########
@@ -0,0 +1,30 @@
+/*
+ * 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;
+
+/**
+ * Global event attribute filter applied before delivery to all listeners. Determines which {@link
+ * AttributeKey}s from an {@link EventAttributeMap} are safe to pass downstream. Sensitive
+ * attributes are stripped globally so all listeners receive only sanitized events by default.
+ */
+public interface EventAttributeFilter {
Review Comment:
Thanks for adding sanitizer interface. Do we actually need a separate EventAttributeFilter interface? Right now it has one impl (DefaultEventAttributeFilter) and one consumer (DefaultEventSanitizer), the dispatcher only ever talks to EventSanitizer. I'd lean toward folding it into DefaultEventSanitizer:
- The common customization("deny extra keys") is already handled by polaris.event-listener.denylisted-attributes, no bean swap needed. The separate filter bean only buys swapping the denial policy wholesale (e.g. allowlist), which is just as easily done by replacing the one EventSanitizer bean.
- Deny and derive aren't cleanly separable anyway: extractDerivedAttributes reaches around the filter to read the denied CATALOG object so it can derive CATALOG_NAME. The sanitizer already has to know what the filter drops, so the two belong together.
- The predicate is a one-liner (!denylist.contains(key)); the real logic is the config-driven denylist resolution, which is already the static resolveAdditionalDenylist(...) helper and can stay static.
One caveat: if the allowlist-vs-denylist question from the dev@ thread is still open and you'd want operators to pick the posture, a swappable policy has some value, but even then I'd make it the single EventSanitizer bean rather than keep two interfaces.
##########
runtime/service/src/main/java/org/apache/polaris/service/events/EventSanitizer.java:
##########
@@ -0,0 +1,38 @@
+/*
+ * 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;
+
+/**
+ * Global, dispatcher-enforced sanitizer for events. Invoked exactly once at the choke point in
Review Comment:
Doc nit: `deliverEvent` is called per-listener, so `sanitize()` actually runs (and re-clones) once per listener, not "exactly once at the choke point." Harmless but the javadoc overstates it.
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -80,8 +82,13 @@ public void onStartup(@Observes StartupEvent event) {
private void deliverEvent(
PolarisEvent event, String listenerName, PolarisEventListener listener) {
LOGGER.debug("Delivering {} event to listener '{}' ({})", event.type(), listenerName, listener);
+ // TODO: Add an opt-in mechanism (e.g., a marker interface or annotation) for specialized
Review Comment:
I believe some audit use cases need sensitive data, including Principal, and others. Both maker interface and annotation should work. Example of maker interface:
```
/** Opt-in: listeners implementing this receive the raw, unsanitized event. */
public interface RawEventAccess {}
A listener opts in by implementing it:
@Identifier("forensic-audit")
public class ForensicAuditListener implements PolarisEventListener, RawEventAccess {
@Override public void onEvent(PolarisEvent event) { /* sees raw event */ }
}
Dispatcher check in deliverEvent:
PolarisEvent toDeliver =
(listener instanceof RawEventAccess) ? event : eventSanitizer.sanitize(event);
listener.onEvent(toDeliver);
```
I think we are not plan to apply different sanitizer to different event listener, right? @sririshindra, @nandorKollar . At least the current impl. can't support it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor persistence event listener to a universal registry pattern with strict allowlist serialization [polaris]
Posted by "adutra (via GitHub)" <gi...@apache.org>.
adutra commented on code in PR #4225:
URL: https://github.com/apache/polaris/pull/4225#discussion_r3126381578
##########
runtime/service/src/main/java/org/apache/polaris/service/events/listeners/DefaultPersistenceEventHandler.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * 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.listeners;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Objects;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.rest.requests.RenameTableRequest;
+import org.apache.polaris.core.admin.model.Catalog;
+import org.apache.polaris.core.admin.model.Principal;
+import org.apache.polaris.core.auth.PolarisPrincipal;
+import org.apache.polaris.core.entity.PolarisEvent.ResourceType;
+import org.apache.polaris.service.events.EventAttributeMap;
+import org.apache.polaris.service.events.EventAttributes;
+import org.apache.polaris.service.events.PolarisEvent;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+final class DefaultPersistenceEventHandler implements PersistenceEventHandler {
+
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(DefaultPersistenceEventHandler.class);
+ private static final String SERIALIZATION_ERROR_PAYLOAD = "{\"error\":\"Serialization failed\"}";
+
+ static final String ALL_CATALOGS_SCOPE = "_all_catalogs";
+ static final String ROOT_NAMESPACE_SCOPE = "_root_namespace";
+ static final String TABLE_SCOPE = "_table_scope";
+ static final String VIEW_SCOPE = "_view_scope";
+
+ private final PersistenceAttributeSecurityFilter securityFilter;
+ private final ObjectMapper objectMapper;
+
+ DefaultPersistenceEventHandler() {
+ this(new PersistenceAttributeSecurityFilter(), new ObjectMapper());
+ }
+
+ DefaultPersistenceEventHandler(
+ PersistenceAttributeSecurityFilter securityFilter, ObjectMapper objectMapper) {
+ this.securityFilter = Objects.requireNonNull(securityFilter, "securityFilter");
+ this.objectMapper = Objects.requireNonNull(objectMapper, "objectMapper");
+ }
+
+ @Override
+ public org.apache.polaris.core.entity.PolarisEvent toPersistenceEvent(PolarisEvent event) {
+ EventAttributeMap attributes = event.attributes();
+ String catalogName = resolveCatalogName(attributes);
+ ResourceType resourceType = resolveResourceType(event, attributes);
+
+ org.apache.polaris.core.entity.PolarisEvent persistenceEvent =
+ new org.apache.polaris.core.entity.PolarisEvent(
+ catalogName,
+ event.metadata().eventId().toString(),
+ event.metadata().requestId().orElse(null),
+ event.type().name(),
+ event.metadata().timestamp().toEpochMilli(),
+ event.metadata().user().map(PolarisPrincipal::getName).orElse(null),
+ resourceType,
+ resolveResourceIdentifier(resourceType, attributes, catalogName));
+
+ Map<String, String> additionalProperties =
+ new LinkedHashMap<>(securityFilter.filterAllowlistedAttributes(attributes));
+ additionalProperties.putAll(
+ securityFilter.filterAllowlistedTelemetryContext(event.metadata().openTelemetryContext()));
+ if (!additionalProperties.isEmpty()) {
+ persistenceEvent.setAdditionalProperties(
+ serializeAdditionalProperties(event, additionalProperties));
+ }
+
+ return persistenceEvent;
+ }
+
+ private String serializeAdditionalProperties(
+ PolarisEvent event, Map<String, String> additionalProperties) {
+ try {
+ return objectMapper.writeValueAsString(additionalProperties);
+ } catch (JsonProcessingException e) {
+ LOGGER.warn(
+ "Failed to serialize additional properties for eventType={} eventId={}; persisting fallback payload",
+ event.type(),
+ event.metadata().eventId());
+ LOGGER.debug(
+ "Serialization failure details for eventType={} eventId={}",
+ event.type(),
+ event.metadata().eventId(),
+ e);
+ return SERIALIZATION_ERROR_PAYLOAD;
+ }
+ }
+
+ private static String resolveCatalogName(EventAttributeMap attributes) {
+ return attributes
+ .get(EventAttributes.CATALOG_NAME)
+ .or(() -> attributes.get(EventAttributes.CATALOG).map(Catalog::getName))
+ .orElse(ALL_CATALOGS_SCOPE);
+ }
+
+ private static ResourceType resolveResourceType(
+ PolarisEvent event, EventAttributeMap attributes) {
+ return switch (event.type().category()) {
+ case CATALOG -> ResourceType.CATALOG;
+ case NAMESPACE -> ResourceType.NAMESPACE;
+ case TABLE -> ResourceType.TABLE;
+ case VIEW -> ResourceType.VIEW;
+ // Persistence resource type schema supports CATALOG/NAMESPACE/TABLE/VIEW only.
Review Comment:
Why? I think all event types should be persisted.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Overhaul event persistence with universal routing and global secure-by-default sanitization [polaris]
Posted by "sririshindra (via GitHub)" <gi...@apache.org>.
sririshindra commented on PR #4225:
URL: https://github.com/apache/polaris/pull/4225#issuecomment-4662874541
> > Is there any change that you would like me to do in this PR itself and not a follow-up? If not, can you approve this PR so that it can get merged and I will address your comments in a follow-up? Does that work?
>
> Could you please create the GitHub issue for the `REALM_SCOPED` sentinel value, then tackle just this nit:
>
> [#4225 (comment)](https://github.com/apache/polaris/pull/4225#discussion_r3379625418)
>
> After that I think we are good to merge.
Thanks @adutra , I have filed https://github.com/apache/polaris/issues/4674 and also addressed the nit. I will file a follow up PR for the issue and rest of the comment once this PR gets merged.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor persistence event listener to a universal registry pattern with strict allowlist serialization [polaris]
Posted by "sririshindra (via GitHub)" <gi...@apache.org>.
sririshindra commented on code in PR #4225:
URL: https://github.com/apache/polaris/pull/4225#discussion_r3231018644
##########
runtime/service/src/main/java/org/apache/polaris/service/events/listeners/PolarisPersistenceEventListener.java:
##########
@@ -19,75 +19,31 @@
package org.apache.polaris.service.events.listeners;
-import com.google.common.collect.ImmutableMap;
import java.util.Map;
-import org.apache.iceberg.TableMetadata;
-import org.apache.iceberg.TableMetadataParser;
-import org.apache.iceberg.catalog.Namespace;
-import org.apache.iceberg.catalog.TableIdentifier;
-import org.apache.iceberg.rest.responses.LoadTableResponse;
-import org.apache.polaris.core.admin.model.Catalog;
-import org.apache.polaris.core.auth.PolarisPrincipal;
-import org.apache.polaris.service.events.EventAttributes;
+import java.util.Objects;
import org.apache.polaris.service.events.PolarisEvent;
+import org.apache.polaris.service.events.PolarisEventType;
public abstract class PolarisPersistenceEventListener implements PolarisEventListener {
- @Override
- public void onEvent(PolarisEvent event) {
- switch (event.type()) {
- case AFTER_CREATE_TABLE -> handleAfterCreateTable(event);
- case AFTER_CREATE_CATALOG -> handleAfterCreateCatalog(event);
- default -> {
- // Other events not handled by this listener
- }
- }
- }
+ private final PersistenceEventHandler defaultHandler;
+ private final Map<PolarisEventType, PersistenceEventHandler> eventTypeHandlers;
- private void handleAfterCreateTable(PolarisEvent event) {
- LoadTableResponse loadTableResponse =
- event.attributes().getRequired(EventAttributes.LOAD_TABLE_RESPONSE);
- TableMetadata tableMetadata = loadTableResponse.tableMetadata();
- String catalogName = event.attributes().getRequired(EventAttributes.CATALOG_NAME);
- Namespace namespace = event.attributes().getRequired(EventAttributes.NAMESPACE);
- String tableName = event.attributes().getRequired(EventAttributes.TABLE_NAME);
+ protected PolarisPersistenceEventListener() {
Review Comment:
Removed the handler registry entirely. The universal approach (single listener handling all events via category() switch + allowlist) eliminates the need for per-event-type handlers. The filter and pruner are now CDI beans (@ApplicationScoped @DefaultBean), so operators can provide alternative implementations via standard CDI mechanisms.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Overhaul event persistence with universal routing and global secure-by-default sanitization [polaris]
Posted by "flyrain (via GitHub)" <gi...@apache.org>.
flyrain commented on code in PR #4225:
URL: https://github.com/apache/polaris/pull/4225#discussion_r3262281270
##########
runtime/service/src/main/java/org/apache/polaris/service/events/listeners/PolarisPersistenceEventListener.java:
##########
@@ -19,38 +19,37 @@
package org.apache.polaris.service.events.listeners;
-import com.google.common.collect.ImmutableMap;
+import jakarta.inject.Inject;
+import java.util.LinkedHashMap;
import java.util.Map;
-import org.apache.iceberg.TableMetadata;
-import org.apache.iceberg.TableMetadataParser;
+import java.util.Optional;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
-import org.apache.iceberg.rest.responses.LoadTableResponse;
-import org.apache.polaris.core.admin.model.Catalog;
+import org.apache.iceberg.rest.requests.CreateTableRequest;
+import org.apache.iceberg.rest.requests.RegisterTableRequest;
import org.apache.polaris.core.auth.PolarisPrincipal;
+import org.apache.polaris.service.events.EventAttributeMap;
import org.apache.polaris.service.events.EventAttributes;
+import org.apache.polaris.service.events.EventPayloadPruner;
import org.apache.polaris.service.events.PolarisEvent;
+import org.apache.polaris.service.events.PolarisEventType;
public abstract class PolarisPersistenceEventListener implements PolarisEventListener {
- @Override
- public void onEvent(PolarisEvent event) {
- switch (event.type()) {
- case AFTER_CREATE_TABLE -> handleAfterCreateTable(event);
- case AFTER_CREATE_CATALOG -> handleAfterCreateCatalog(event);
- default -> {
- // Other events not handled by this listener
- }
- }
+ @Inject EventPayloadPruner payloadPruner;
+
+ protected PolarisPersistenceEventListener() {}
+
+ PolarisPersistenceEventListener(EventPayloadPruner payloadPruner) {
+ this.payloadPruner = payloadPruner;
}
- private void handleAfterCreateTable(PolarisEvent event) {
- LoadTableResponse loadTableResponse =
- event.attributes().getRequired(EventAttributes.LOAD_TABLE_RESPONSE);
- TableMetadata tableMetadata = loadTableResponse.tableMetadata();
- String catalogName = event.attributes().getRequired(EventAttributes.CATALOG_NAME);
- Namespace namespace = event.attributes().getRequired(EventAttributes.NAMESPACE);
- String tableName = event.attributes().getRequired(EventAttributes.TABLE_NAME);
+ @Override
+ public void onEvent(PolarisEvent event) {
+ String catalogName = resolveCatalogName(event);
Review Comment:
It's confusing that a catalog name now can be a realm constant, like `__realm__`. We should avoid that if possible. Is that because catalog id couldn't be null in DB schema? Can we remove the constraint so that catalog id could be null in case of realm-level events? cc @adnanhemani
##########
runtime/service/src/main/java/org/apache/polaris/service/events/listeners/DefaultEventPayloadPruner.java:
##########
@@ -0,0 +1,91 @@
+/*
+ * 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.listeners;
+
+import io.quarkus.arc.DefaultBean;
+import jakarta.enterprise.context.ApplicationScoped;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.rest.requests.RenameTableRequest;
+import org.apache.iceberg.rest.responses.LoadTableResponse;
+import org.apache.polaris.service.events.AttributeKey;
+import org.apache.polaris.service.events.EventAttributes;
+import org.apache.polaris.service.events.EventPayloadPruner;
+
+@ApplicationScoped
+@DefaultBean
Review Comment:
Pruning is explicitly per-listener by design, the JavaDoc on EventPayloadPruner says so:
> While EventAttributeFilter decides which keys pass globally, this interface decides how a passing value is serialized, and is applied per-listener as a serialization concern.
So nothing in the framework enforces that any listener uses EventPayloadPruner at all. A future CloudWatch listener could ignore the injected pruner, write its own logic, and the dispatcher wouldn't know or care. The "contract" suggested by the interface + CDI bean shape doesn't actually exist, it's just a helper that one listener happens to inject.
That makes the current shape misleading on two fronts:
1. CDI suggests a global hook (like EventAttributeFilter, which is enforced in deliverEvent()). But pruning isn't a global hook. Reading the code, you'd reasonably assume swapping the bean changes pruning everywhere, it doesn't.
2. One listener, one impl, no contract = the indirection earns nothing. A static prune(key, value) util on the persistence listener (or a sibling helper class) would do exactly the same job with less ceremony.
The two coherent designs are:
- Make it global. Move pruning into PolarisEventListeners.deliverEvent() next to sanitize(). Then the CDI bean shape makes sense because every listener gets pruned events.
- Keep it per-listener. Drop the interface + bean and inline it as a util / private method on PolarisPersistenceEventListener. The interface only earns its keep when there's a second impl that needs to be selected.
##########
runtime/service/src/main/java/org/apache/polaris/service/events/EventAttributeFilter.java:
##########
@@ -0,0 +1,30 @@
+/*
+ * 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;
+
+/**
+ * Global event attribute filter applied before delivery to all listeners. Determines which {@link
+ * AttributeKey}s from an {@link EventAttributeMap} are safe to pass downstream. Sensitive
+ * attributes are stripped globally so all listeners receive only sanitized events by default.
+ */
+public interface EventAttributeFilter {
Review Comment:
`EventAttributeFilter` is too narrow for what's running at the choke point
`PolarisEventListeners.sanitize()` does two things:
1. `attributeFilter.isAllowed(key)` per attribute
2. `extractDerivedAttributes()` (CATALOG โ CATALOG_NAME)
Both are sanitization concerns that run at the same point for the same reason, but one is an injected bean and the other is a private static method. An operator who replaces `EventAttributeFilter` doesn't get to influence derivation, that policy is locked inside the dispatcher.
Consider an `EventSanitizer` bean as the global, enforced abstraction:
```java
interface EventSanitizer {
PolarisEvent sanitize(PolarisEvent event);
}
```
A `DefaultEventSanitizer` holds the denylist (or a sub-EventAttributeFilter if you want to keep that extension point) plus the derivation logic. The dispatcher just calls `sanitizer.sanitize(event)`. This also gives the future "opt-in raw-event escape hatch" (TODO at PolarisEventListeners.java:85) a natural home, it becomes a strategy choice inside the sanitizer rather than another branch in deliverEvent.
##########
runtime/service/src/main/java/org/apache/polaris/service/events/listeners/PolarisPersistenceEventListener.java:
##########
@@ -60,34 +59,187 @@ private void handleAfterCreateTable(PolarisEvent event) {
event.type().name(),
event.metadata().timestamp().toEpochMilli(),
event.metadata().user().map(PolarisPrincipal::getName).orElse(null),
- org.apache.polaris.core.entity.PolarisEvent.ResourceType.TABLE,
- TableIdentifier.of(namespace, tableName).toString());
- var additionalParameters =
- ImmutableMap.<String, String>builder()
- .put("table-uuid", tableMetadata.uuid())
- .put("metadata", TableMetadataParser.toJson(tableMetadata));
- additionalParameters.putAll(event.metadata().openTelemetryContext());
- polarisEvent.setAdditionalProperties(additionalParameters.build());
+ resourceType,
+ resourceIdentifier);
+
+ Map<String, String> additionalProperties = buildAdditionalProperties(event);
+ if (!additionalProperties.isEmpty()) {
+ polarisEvent.setAdditionalProperties(additionalProperties);
+ }
+
processEvent(event.metadata().realmId(), polarisEvent);
}
- private void handleAfterCreateCatalog(PolarisEvent event) {
- Catalog catalog = event.attributes().getRequired(EventAttributes.CATALOG);
- org.apache.polaris.core.entity.PolarisEvent polarisEvent =
- new org.apache.polaris.core.entity.PolarisEvent(
- catalog.getName(),
- event.metadata().eventId().toString(),
- event.metadata().requestId().orElse(null),
- event.type().name(),
- event.metadata().timestamp().toEpochMilli(),
- event.metadata().user().map(PolarisPrincipal::getName).orElse(null),
- org.apache.polaris.core.entity.PolarisEvent.ResourceType.CATALOG,
- catalog.getName());
- Map<String, String> openTelemetryContext = event.metadata().openTelemetryContext();
- if (!openTelemetryContext.isEmpty()) {
- polarisEvent.setAdditionalProperties(openTelemetryContext);
+ private static String resolveCatalogName(PolarisEvent event) {
+ return event
+ .attributes()
+ .get(EventAttributes.CATALOG_NAME)
+ .orElse(org.apache.polaris.core.entity.PolarisEvent.REALM_SCOPED);
+ }
+
+ private static org.apache.polaris.core.entity.PolarisEvent.ResourceType resolveResourceType(
+ PolarisEventType eventType) {
+ return switch (eventType.category()) {
+ case TABLE, GENERIC_TABLE -> org.apache.polaris.core.entity.PolarisEvent.ResourceType.TABLE;
+ case VIEW -> org.apache.polaris.core.entity.PolarisEvent.ResourceType.VIEW;
+ case NAMESPACE -> org.apache.polaris.core.entity.PolarisEvent.ResourceType.NAMESPACE;
+ case CATALOG -> org.apache.polaris.core.entity.PolarisEvent.ResourceType.CATALOG;
+ default -> org.apache.polaris.core.entity.PolarisEvent.ResourceType.REALM;
Review Comment:
The default -> REALM arm in resolveResourceType is a silent catch-all. It lumps PRINCIPAL, CATALOG_ROLE, PRINCIPAL_ROLE, POLICY, CREDENTIAL, TRANSACTION, NOTIFICATION, CONFIG, etc. all under ResourceType.REALM. That's lossy for audit queries (a policy change isn't realm-scoped) and silently routes any future Category value to REALM with no compile error. I'd suggest to
make the switch exhaustive (no default) and explicitly map only the genuinely realm-scoped categories to REALM.
##########
runtime/service/src/main/java/org/apache/polaris/service/events/listeners/PolarisPersistenceEventListener.java:
##########
@@ -19,38 +19,37 @@
package org.apache.polaris.service.events.listeners;
-import com.google.common.collect.ImmutableMap;
+import jakarta.inject.Inject;
+import java.util.LinkedHashMap;
import java.util.Map;
-import org.apache.iceberg.TableMetadata;
-import org.apache.iceberg.TableMetadataParser;
+import java.util.Optional;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
-import org.apache.iceberg.rest.responses.LoadTableResponse;
-import org.apache.polaris.core.admin.model.Catalog;
+import org.apache.iceberg.rest.requests.CreateTableRequest;
+import org.apache.iceberg.rest.requests.RegisterTableRequest;
import org.apache.polaris.core.auth.PolarisPrincipal;
+import org.apache.polaris.service.events.EventAttributeMap;
import org.apache.polaris.service.events.EventAttributes;
+import org.apache.polaris.service.events.EventPayloadPruner;
import org.apache.polaris.service.events.PolarisEvent;
+import org.apache.polaris.service.events.PolarisEventType;
public abstract class PolarisPersistenceEventListener implements PolarisEventListener {
- @Override
- public void onEvent(PolarisEvent event) {
- switch (event.type()) {
- case AFTER_CREATE_TABLE -> handleAfterCreateTable(event);
- case AFTER_CREATE_CATALOG -> handleAfterCreateCatalog(event);
- default -> {
- // Other events not handled by this listener
- }
- }
+ @Inject EventPayloadPruner payloadPruner;
+
+ protected PolarisPersistenceEventListener() {}
+
+ PolarisPersistenceEventListener(EventPayloadPruner payloadPruner) {
+ this.payloadPruner = payloadPruner;
}
- private void handleAfterCreateTable(PolarisEvent event) {
- LoadTableResponse loadTableResponse =
- event.attributes().getRequired(EventAttributes.LOAD_TABLE_RESPONSE);
- TableMetadata tableMetadata = loadTableResponse.tableMetadata();
- String catalogName = event.attributes().getRequired(EventAttributes.CATALOG_NAME);
- Namespace namespace = event.attributes().getRequired(EventAttributes.NAMESPACE);
- String tableName = event.attributes().getRequired(EventAttributes.TABLE_NAME);
+ @Override
+ public void onEvent(PolarisEvent event) {
+ String catalogName = resolveCatalogName(event);
+ org.apache.polaris.core.entity.PolarisEvent.ResourceType resourceType =
Review Comment:
nit: we can import it to avoid the full path reference for `ResourceType`
##########
runtime/service/src/main/java/org/apache/polaris/service/events/listeners/DefaultPersistenceEventHandler.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * 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.listeners;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Objects;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.rest.requests.RenameTableRequest;
+import org.apache.polaris.core.admin.model.Catalog;
+import org.apache.polaris.core.admin.model.Principal;
+import org.apache.polaris.core.auth.PolarisPrincipal;
+import org.apache.polaris.core.entity.PolarisEvent.ResourceType;
+import org.apache.polaris.service.events.EventAttributeMap;
+import org.apache.polaris.service.events.EventAttributes;
+import org.apache.polaris.service.events.PolarisEvent;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+final class DefaultPersistenceEventHandler implements PersistenceEventHandler {
+
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(DefaultPersistenceEventHandler.class);
+ private static final String SERIALIZATION_ERROR_PAYLOAD = "{\"error\":\"Serialization failed\"}";
+
+ static final String ALL_CATALOGS_SCOPE = "_all_catalogs";
+ static final String ROOT_NAMESPACE_SCOPE = "_root_namespace";
+ static final String TABLE_SCOPE = "_table_scope";
Review Comment:
These sentinels (`_table_scope`, `_view_scope`, `_all_catalogs`, `_root_namespace`) get written directly into the `catalogId` / `resourceIdentifier` columns. Works, but overloads those fields, downstream queries now need to filter sentinels out. Consider a dedicated scope `ResourceType` (e.g. `ROOT`/`SCOPE`) or a nullable column; at minimum document the convention on `org.apache.polaris.core.entity.PolarisEvent`.
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -80,8 +82,13 @@ public void onStartup(@Observes StartupEvent event) {
private void deliverEvent(
PolarisEvent event, String listenerName, PolarisEventListener listener) {
LOGGER.debug("Delivering {} event to listener '{}' ({})", event.type(), listenerName, listener);
+ // TODO: Add an opt-in mechanism (e.g., a marker interface or annotation) for specialized
Review Comment:
How will "opt-in" work for custom listener? Can you share more details? We need to think it through to avoid a big redesign.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Overhaul event persistence with universal routing and global secure-by-default sanitization [polaris]
Posted by "nandorKollar (via GitHub)" <gi...@apache.org>.
nandorKollar commented on PR #4225:
URL: https://github.com/apache/polaris/pull/4225#issuecomment-4638057480
With this sanitisation logic, we can filter out AttributeKeys but can't filter out a certain property of an object right? So for example, we can define that we don't allow event listeners to receive `UpdatePrincipalRequest` (with `update_catalog_request` key), but we can filter out just some properties of this request right? I think it is fine, just would like to understand the limitations, and I'm wondering that the scenario is in fact relevant or not at all. The overall approach looks good to me.
I think we should also mention this new feature in the release notes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Overhaul event persistence with universal routing and global secure-by-default sanitization [polaris]
Posted by "nandorKollar (via GitHub)" <gi...@apache.org>.
nandorKollar commented on code in PR #4225:
URL: https://github.com/apache/polaris/pull/4225#discussion_r3298921861
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -80,8 +82,13 @@ public void onStartup(@Observes StartupEvent event) {
private void deliverEvent(
PolarisEvent event, String listenerName, PolarisEventListener listener) {
LOGGER.debug("Delivering {} event to listener '{}' ({})", event.type(), listenerName, listener);
+ // TODO: Add an opt-in mechanism (e.g., a marker interface or annotation) for specialized
Review Comment:
I think either a marker interface, or an extra method on `PolarisEventListener` with a default implement could address this. Something like this:
```
default boolean acceptUnsanitizedEvent() {
return false;
}
```
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -80,8 +82,13 @@ public void onStartup(@Observes StartupEvent event) {
private void deliverEvent(
PolarisEvent event, String listenerName, PolarisEventListener listener) {
LOGGER.debug("Delivering {} event to listener '{}' ({})", event.type(), listenerName, listener);
+ // TODO: Add an opt-in mechanism (e.g., a marker interface or annotation) for specialized
Review Comment:
I think either a marker interface, or an extra method on `PolarisEventListener` with a default implement could address this. Something like this:
```
default boolean acceptUnsanitizedEvent() {
return false;
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Overhaul event persistence with universal routing and global secure-by-default sanitization [polaris]
Posted by "flyrain (via GitHub)" <gi...@apache.org>.
flyrain merged PR #4225:
URL: https://github.com/apache/polaris/pull/4225
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor persistence event listener to a universal registry pattern with strict allowlist serialization [polaris]
Posted by "sririshindra (via GitHub)" <gi...@apache.org>.
sririshindra commented on code in PR #4225:
URL: https://github.com/apache/polaris/pull/4225#discussion_r3231028386
##########
runtime/service/src/main/java/org/apache/polaris/service/events/listeners/PersistenceAttributeSecurityFilter.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.listeners;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Set;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableMetadataParser;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.rest.responses.LoadTableResponse;
+import org.apache.polaris.service.events.AttributeKey;
+import org.apache.polaris.service.events.EventAttributeMap;
+import org.apache.polaris.service.events.EventAttributes;
+
+final class PersistenceAttributeSecurityFilter {
+
+ private static final Set<String> ALLOWED_TELEMETRY_CONTEXT_KEYS =
+ Set.of("otel.trace_id", "otel.span_id", "otel.trace_flags", "otel.sampled");
+
+ Map<String, String> filterAllowlistedAttributes(EventAttributeMap attributes) {
+ Map<String, String> safeAttributes = new LinkedHashMap<>();
+ attributes.forEach((key, value) -> addIfAllowlisted(safeAttributes, key, value));
+ return safeAttributes;
+ }
+
+ Map<String, String> filterAllowlistedTelemetryContext(Map<String, String> telemetryContext) {
+ Map<String, String> safeContext = new LinkedHashMap<>();
+ telemetryContext.forEach(
+ (key, value) -> {
+ if (ALLOWED_TELEMETRY_CONTEXT_KEYS.contains(key)) {
+ safeContext.put(key, value);
+ }
+ });
+ return safeContext;
+ }
+
+ @SuppressWarnings("unchecked")
+ private static void addIfAllowlisted(
+ Map<String, String> safeAttributes, AttributeKey<?> key, Object value) {
+ if (EventAttributes.CATALOG_NAME.equals(key)
+ || EventAttributes.TABLE_NAME.equals(key)
+ || EventAttributes.VIEW_NAME.equals(key)
+ || EventAttributes.NAMESPACE_FQN.equals(key)
+ || EventAttributes.PARENT_NAMESPACE_FQN.equals(key)) {
+ safeAttributes.put(key.name(), (String) value);
+ return;
+ }
+
+ if (EventAttributes.NAMESPACE.equals(key)) {
+ safeAttributes.put(key.name(), ((Namespace) value).toString());
+ return;
+ }
+
+ if (EventAttributes.TABLE_IDENTIFIER.equals(key)
+ || EventAttributes.VIEW_IDENTIFIER.equals(key)) {
+ safeAttributes.put(key.name(), ((TableIdentifier) value).toString());
+ return;
+ }
+
+ // TODO: Filter sensitive events and credential-bearing table properties before serializing
+ // table metadata. This and other sensitive event filtering cases will be addressed separately.
+ if (EventAttributes.TABLE_METADATA.equals(key)) {
+ addTableMetadata(safeAttributes, (TableMetadata) value);
+ return;
+ }
+
+ if (EventAttributes.LOAD_TABLE_RESPONSE.equals(key)) {
+ addTableMetadata(safeAttributes, ((LoadTableResponse) value).tableMetadata());
+ return;
+ }
+ }
+
+ private static void addTableMetadata(
+ Map<String, String> safeAttributes, TableMetadata tableMetadata) {
+ if (tableMetadata == null) {
+ return;
+ }
+
+ safeAttributes.put(
+ EventAttributes.TABLE_METADATA.name(), TableMetadataParser.toJson(tableMetadata));
Review Comment:
Fixed. DefaultEventPayloadPruner.pruneTableMetadata() now extracts only: table_uuid, table_location, table_format_version, table_current_snapshot_id, table_schema, table_last_updated_ms. Guaranteed bounded output regardless of snapshot/partition history size.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor persistence event listener to a universal registry pattern with strict allowlist serialization [polaris]
Posted by "sririshindra (via GitHub)" <gi...@apache.org>.
sririshindra commented on code in PR #4225:
URL: https://github.com/apache/polaris/pull/4225#discussion_r3231026782
##########
runtime/service/src/main/java/org/apache/polaris/service/events/listeners/PersistenceAttributeSecurityFilter.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.listeners;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Set;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableMetadataParser;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.rest.responses.LoadTableResponse;
+import org.apache.polaris.service.events.AttributeKey;
+import org.apache.polaris.service.events.EventAttributeMap;
+import org.apache.polaris.service.events.EventAttributes;
+
+final class PersistenceAttributeSecurityFilter {
Review Comment:
Done. Split into two CDI beans:
- DefaultEventAttributeFilter โ allowlist-based, configurable via polaris.event-listener.persistence.allowlisted-attributes
- DefaultEventPayloadPruner โ prunes TableMetadata to a bounded summary (uuid, location, format-version, current-snapshot-id, schema, last-updated-ms โ ~1KB instead of potentially hundreds of MB)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor persistence event listener to strategy-based table handlers and add dedicated table handler classes [polaris]
Posted by "sririshindra (via GitHub)" <gi...@apache.org>.
sririshindra commented on PR #4225:
URL: https://github.com/apache/polaris/pull/4225#issuecomment-4285006935
> #4227
Thanks @youngrae81 . Let me update this PR now that https://github.com/apache/polaris/pull/4064 is merged and we can discuss further.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor persistence event listener to a universal registry pattern with strict allowlist serialization [polaris]
Posted by "sririshindra (via GitHub)" <gi...@apache.org>.
sririshindra commented on PR #4225:
URL: https://github.com/apache/polaris/pull/4225#issuecomment-4292778092
Thanks @nandorKollar, @adutra, @youngrae81, and @obelix74 for the review earlier. I have completely pivoted the architecture to a universally scalable fallback handler backed by a strict security allowlist. This resolves the boilerplate scaling issue, covers all events natively (including Views), and prevents sensitive table properties from leaking into the audit log. Could you please take another look at the PR
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Overhaul event persistence with universal routing and global secure-by-default sanitization [polaris]
Posted by "adutra (via GitHub)" <gi...@apache.org>.
adutra commented on code in PR #4225:
URL: https://github.com/apache/polaris/pull/4225#discussion_r3379179465
##########
polaris-core/src/main/java/org/apache/polaris/core/entity/PolarisEvent.java:
##########
@@ -30,11 +30,30 @@
public class PolarisEvent {
public static final String EMPTY_MAP_STRING = "{}";
+ /**
+ * Sentinel stored in {@link #catalogId} for events that are not scoped to a specific catalog
+ * (e.g., principal, principal-role, or other realm-level operations). Downstream consumers
+ * querying by catalog must filter this sentinel out (or coalesce it to {@code NULL}).
+ *
+ * <p>This is a transitional placeholder: {@code catalogId} is currently {@code NOT NULL} in the
+ * persistence schema, so realm-scoped events cannot store {@code null} directly. A future schema
+ * migration may relax that constraint and let realm-scoped events use {@code NULL} instead.
+ *
+ * <p>TODO: Remove this sentinel once the {@code catalogId} column is made nullable. Realm-scoped
+ * events should store {@code NULL}, not a magic string, so downstream queries can use {@code IS
+ * NULL} instead of filtering out a reserved value.
+ */
+ public static final String REALM_SCOPED = "__realm__";
Review Comment:
Do we have a GitHub issue to track this?
##########
runtime/service/src/main/java/org/apache/polaris/service/events/listeners/PolarisPersistenceEventListener.java:
##########
@@ -60,34 +55,256 @@ private void handleAfterCreateTable(PolarisEvent event) {
event.type().name(),
event.metadata().timestamp().toEpochMilli(),
event.metadata().user().map(PolarisPrincipal::getName).orElse(null),
- org.apache.polaris.core.entity.PolarisEvent.ResourceType.TABLE,
- TableIdentifier.of(namespace, tableName).toString());
- var additionalParameters =
- ImmutableMap.<String, String>builder()
- .put("table-uuid", tableMetadata.uuid())
- .put("metadata", TableMetadataParser.toJson(tableMetadata));
- additionalParameters.putAll(event.metadata().openTelemetryContext());
- polarisEvent.setAdditionalProperties(additionalParameters.build());
+ resourceType,
+ resourceIdentifier);
+
+ Map<String, String> additionalProperties = buildAdditionalProperties(event);
+ if (!additionalProperties.isEmpty()) {
+ polarisEvent.setAdditionalProperties(additionalProperties);
+ }
+
processEvent(event.metadata().realmId(), polarisEvent);
}
- private void handleAfterCreateCatalog(PolarisEvent event) {
- Catalog catalog = event.attributes().getRequired(EventAttributes.CATALOG);
- org.apache.polaris.core.entity.PolarisEvent polarisEvent =
- new org.apache.polaris.core.entity.PolarisEvent(
- catalog.getName(),
- event.metadata().eventId().toString(),
- event.metadata().requestId().orElse(null),
- event.type().name(),
- event.metadata().timestamp().toEpochMilli(),
- event.metadata().user().map(PolarisPrincipal::getName).orElse(null),
- org.apache.polaris.core.entity.PolarisEvent.ResourceType.CATALOG,
- catalog.getName());
- Map<String, String> openTelemetryContext = event.metadata().openTelemetryContext();
- if (!openTelemetryContext.isEmpty()) {
- polarisEvent.setAdditionalProperties(openTelemetryContext);
+ private static String resolveCatalogName(PolarisEvent event) {
+ return event
+ .attributes()
+ .get(EventAttributes.CATALOG_NAME)
+ .orElse(org.apache.polaris.core.entity.PolarisEvent.REALM_SCOPED);
+ }
+
+ /**
+ * Maps an event's category to a {@link ResourceType} for indexing. Exhaustive over all {@link
+ * PolarisEventType.Category} values: any new category added to the enum must be classified here
+ * explicitly. The {@code default} arm is intentionally absent so that adding a category produces
+ * a compile-time error rather than silently routing to {@link ResourceType#REALM}.
+ */
+ private static ResourceType resolveResourceType(PolarisEventType eventType) {
+ return switch (eventType.category()) {
+ case TABLE, GENERIC_TABLE -> ResourceType.TABLE;
+ case VIEW -> ResourceType.VIEW;
+ case NAMESPACE -> ResourceType.NAMESPACE;
+ case CATALOG -> ResourceType.CATALOG;
+ case POLICY,
+ PRINCIPAL,
+ PRINCIPAL_ROLE,
+ CATALOG_ROLE,
+ CREDENTIAL,
+ TRANSACTION,
+ NOTIFICATION,
+ CONFIG,
+ TASK_EXECUTION,
+ RATE_LIMITING ->
+ ResourceType.REALM;
+ };
+ }
+
+ private static String resolveResourceIdentifier(
+ PolarisEvent event, ResourceType resourceType, String catalogName) {
+ return switch (resourceType) {
+ case TABLE -> resolveTableResourceIdentifier(event, catalogName);
+ case VIEW -> resolveViewResourceIdentifier(event, catalogName);
+ case NAMESPACE -> resolveNamespaceResourceIdentifier(event, catalogName);
+ case CATALOG -> resolveCatalogResourceIdentifier(event, catalogName);
+ case REALM -> resolveRealmResourceIdentifier(event);
+ };
+ }
+
+ private static String resolveTableResourceIdentifier(PolarisEvent event, String catalogName) {
+ EventAttributeMap attributes = event.attributes();
+
+ Optional<String> identifierFromTableAttribute =
+ attributes.get(EventAttributes.TABLE_IDENTIFIER).map(TableIdentifier::toString);
+ if (identifierFromTableAttribute.isPresent()) {
+ return identifierFromTableAttribute.get();
}
- processEvent(event.metadata().realmId(), polarisEvent);
+
+ Optional<String> renameIdentifier = resolveRenameIdentifier(event);
+ if (renameIdentifier.isPresent()) {
+ return renameIdentifier.get();
+ }
+
+ Optional<String> namespaceAndTableName = resolveNamespaceAndTableName(attributes);
+ if (namespaceAndTableName.isPresent()) {
+ return namespaceAndTableName.get();
+ }
+
+ Optional<String> namespaceAndCreateTableName = resolveCreateTableIdentifier(attributes);
+ if (namespaceAndCreateTableName.isPresent()) {
+ return namespaceAndCreateTableName.get();
+ }
+
+ Optional<String> namespaceAndRegisterTableName = resolveRegisterTableIdentifier(attributes);
+ if (namespaceAndRegisterTableName.isPresent()) {
+ return namespaceAndRegisterTableName.get();
+ }
+
+ Optional<String> namespaceIdentifier =
+ attributes.get(EventAttributes.NAMESPACE).map(Namespace::toString);
+ if (namespaceIdentifier.isPresent()) {
+ return namespaceIdentifier.get();
+ }
+
+ return attributes
+ .get(EventAttributes.TABLE_NAME)
+ .orElseGet(() -> fallbackResourceIdentifier(event, catalogName));
+ }
+
+ private static String resolveViewResourceIdentifier(PolarisEvent event, String catalogName) {
+ EventAttributeMap attributes = event.attributes();
+ return attributes
+ .get(EventAttributes.VIEW_IDENTIFIER)
+ .map(TableIdentifier::toString)
+ .or(() -> resolveRenameIdentifier(event))
+ .or(
+ () ->
+ attributes
+ .get(EventAttributes.NAMESPACE)
+ .flatMap(
+ namespace ->
+ attributes
+ .get(EventAttributes.VIEW_NAME)
+ .map(
+ viewName ->
+ TableIdentifier.of(namespace, viewName).toString())))
+ .or(() -> attributes.get(EventAttributes.VIEW_NAME))
+ .orElseGet(() -> fallbackResourceIdentifier(event, catalogName));
+ }
+
+ private static Optional<String> resolveRenameIdentifier(PolarisEvent event) {
+ return event
+ .attributes()
+ .get(EventAttributes.RENAME_TABLE_REQUEST)
+ .map(
+ request ->
+ event.type().name().startsWith("AFTER_")
+ ? request.destination().toString()
+ : request.source().toString());
+ }
+
+ private static Optional<String> resolveNamespaceAndTableName(EventAttributeMap attributes) {
+ return attributes
+ .get(EventAttributes.NAMESPACE)
+ .flatMap(
+ namespace ->
+ attributes
+ .get(EventAttributes.TABLE_NAME)
+ .map(tableName -> TableIdentifier.of(namespace, tableName).toString()));
+ }
+
+ private static Optional<String> resolveCreateTableIdentifier(EventAttributeMap attributes) {
+ return attributes
+ .get(EventAttributes.NAMESPACE)
+ .flatMap(
+ namespace ->
+ attributes
+ .get(EventAttributes.CREATE_TABLE_REQUEST)
+ .map(CreateTableRequest::name)
+ .map(tableName -> TableIdentifier.of(namespace, tableName).toString()));
+ }
+
+ private static Optional<String> resolveRegisterTableIdentifier(EventAttributeMap attributes) {
+ return attributes
+ .get(EventAttributes.NAMESPACE)
+ .flatMap(
+ namespace ->
+ attributes
+ .get(EventAttributes.REGISTER_TABLE_REQUEST)
+ .map(RegisterTableRequest::name)
+ .map(tableName -> TableIdentifier.of(namespace, tableName).toString()));
+ }
+
+ private static String resolveNamespaceResourceIdentifier(PolarisEvent event, String catalogName) {
+ EventAttributeMap attributes = event.attributes();
+ return attributes
+ .get(EventAttributes.NAMESPACE)
+ .map(Namespace::toString)
+ .or(() -> attributes.get(EventAttributes.NAMESPACE_FQN))
+ .or(() -> attributes.get(EventAttributes.PARENT_NAMESPACE_FQN))
+ .orElseGet(() -> fallbackResourceIdentifier(event, catalogName));
+ }
+
+ private static String resolveCatalogResourceIdentifier(PolarisEvent event, String catalogName) {
+ return event
+ .attributes()
+ .get(EventAttributes.CATALOG_NAME)
+ .orElseGet(() -> fallbackResourceIdentifier(event, catalogName));
+ }
+
+ private static String resolveRealmResourceIdentifier(PolarisEvent event) {
+ return event.type().name();
+ }
+
+ private static String fallbackResourceIdentifier(PolarisEvent event, String catalogName) {
+ if (!org.apache.polaris.core.entity.PolarisEvent.REALM_SCOPED.equals(catalogName)) {
+ return catalogName;
+ }
+ return event.type().name();
+ }
+
+ private static Map<String, String> buildAdditionalProperties(PolarisEvent event) {
+ Map<String, String> additionalProperties =
+ new LinkedHashMap<>(event.metadata().openTelemetryContext());
+ event.attributes().forEach((key, value) -> additionalProperties.putAll(prune(key, value)));
+ return additionalProperties;
+ }
+
+ /**
+ * Reduces an attribute value to a bounded string representation suitable for storage in the
+ * persistence event row. Inlined here (rather than exposed as a swappable bean) because pruning
+ * is a serialization concern of this listener only โ other listeners are free to render the same
+ * value differently.
Review Comment:
Although, this is probably a shared concern among listeners. A Kafka listener probably doesn't want to insert a 1GB record in a topic because of a huge table metadata object.
It's fine to leave this here for now, but I'd like to think about a configurable, listener-agnostic pruning layer. See my other comments about how this could be turned into a sanitizer.
##########
runtime/service/src/main/java/org/apache/polaris/service/events/listeners/RawEventAccess.java:
##########
@@ -0,0 +1,31 @@
+/*
+ * 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.listeners;
+
+/**
+ * Marker interface signalling that a {@link PolarisEventListener} requires the raw, unsanitized
+ * event. Listeners that implement this interface receive the original event from {@link
+ * org.apache.polaris.service.events.PolarisEventListeners#deliverEvent}, bypassing the global
+ * {@link org.apache.polaris.service.events.EventSanitizer}.
+ *
+ * <p>This is an explicit security opt-out. Implement only when the listener has a documented need
+ * for sensitive attributes (for example, a forensic audit sink that records principal credentials
+ * or raw catalog configuration). All other listeners receive the sanitized event by default.
+ */
+public interface RawEventAccess {}
Review Comment:
I am a bit against this interface tbh. To opt out (or in) from security, you need to rebuild your listener and re-deploy your application. Moreover, what if a listener wants to keep deny-listing on, but table metadata pruning off? One single interface will not cover that.
I would prefer instead a configuration-based approach.
For instance, we could expand your `EventSanitizer` interface, which currently focuses on deny-listing attributes, to encompass a whole set of sanitization procedures โ each concrete implementation would be an `@ApplicationScoped` bean:
- `DenyListEventSanitizer` : does what `DefaultEventSanitizer` does today
- `TableMetadataSanitizer`: prunes the `TABLE_METADATA` attribute
- `DerivedAttributeSanitizer` : enriches the event with derived attributes (e.g. extract the target entity ID from a catalog or table-metadata attribute)
- etc.
Each sanitizer would be independently configurable, for example:
```properties
polaris.event-listener.sanitizer.deny-list.attributes=attr1,attr2,attr3
polaris.event-listener.sanitizer.table-metadata.max-payload-size=10Mb
polaris.event-listener.sanitizer.table-metadata.include-schema=false
```
Then each listener would declare the sanitizers it wants, pretty much like it declares today which event types and categories it wants:
```properties
polaris.event-listener.<name>.enabled-event-types=...
polaris.event-listener.<name>.enabled-event-categories=...
polaris.event-listener.<name>.enabled-event-sanitizers=deny-list,table-metadata,...
```
Then finally, Polaris would leverage CDI to discover all the sanitizer beans available, and would leverage the event bus to create a pipeline where the event would be first dispatched to intermediate addresses, where it would pass from sanitizer to sanitizer, then forwarded to its final address where it would be handed off to the listener.
##########
runtime/service/src/main/java/org/apache/polaris/service/events/DefaultEventSanitizer.java:
##########
@@ -0,0 +1,127 @@
+/*
+ * 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 io.quarkus.arc.DefaultBean;
+import jakarta.annotation.PostConstruct;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Set;
+import org.apache.polaris.core.admin.model.Catalog;
+
+/**
+ * Default implementation of {@link EventSanitizer}. Drops attributes whose keys are on the denylist
+ * and derives safe attributes (e.g., {@code CATALOG_NAME} extracted from a denied {@code CATALOG}
+ * object) from the originals. The built-in denylist covers attributes that carry credentials or
+ * internal secrets and may be extended via {@code polaris.event-listener.denylisted-attributes}.
+ *
+ * <p>Operators who need a different denial or derivation policy should replace this bean.
+ */
+@ApplicationScoped
+@DefaultBean
+public class DefaultEventSanitizer implements EventSanitizer {
+
+ // These carry full domain objects containing credentials or internal secrets:
+ // PRINCIPAL - may contain client IDs, internal identity metadata
+ // UPDATE_PRINCIPAL_REQUEST - credential rotation data, secret changes
+ // CATALOG - includes StorageConfigInfo with cloud credentials (AWS keys, Azure tokens, etc.)
+ // NOTIFICATION_REQUEST - contains metadata file locations and potentially signed URLs
+ static final Set<AttributeKey<?>> DEFAULT_DENYLIST =
+ Set.of(
+ EventAttributes.PRINCIPAL,
+ EventAttributes.UPDATE_PRINCIPAL_REQUEST,
+ EventAttributes.CATALOG,
+ EventAttributes.NOTIFICATION_REQUEST);
+
+ @Inject PolarisEventListenerConfiguration configuration;
+
+ private Set<AttributeKey<?>> denylist;
+
+ DefaultEventSanitizer() {
+ this(DEFAULT_DENYLIST);
+ }
+
+ DefaultEventSanitizer(Set<AttributeKey<?>> denylist) {
Review Comment:
nit: Maybe annotate with `@VisibleForTesting` if you really intended this constructor to be package-private. If you intended instead to make this a constructor for subclasses to customize the deny list, then you should make it protected.
##########
runtime/service/src/main/java/org/apache/polaris/service/events/listeners/PolarisPersistenceEventListener.java:
##########
@@ -60,34 +55,256 @@ private void handleAfterCreateTable(PolarisEvent event) {
event.type().name(),
event.metadata().timestamp().toEpochMilli(),
event.metadata().user().map(PolarisPrincipal::getName).orElse(null),
- org.apache.polaris.core.entity.PolarisEvent.ResourceType.TABLE,
- TableIdentifier.of(namespace, tableName).toString());
- var additionalParameters =
- ImmutableMap.<String, String>builder()
- .put("table-uuid", tableMetadata.uuid())
- .put("metadata", TableMetadataParser.toJson(tableMetadata));
- additionalParameters.putAll(event.metadata().openTelemetryContext());
- polarisEvent.setAdditionalProperties(additionalParameters.build());
+ resourceType,
+ resourceIdentifier);
+
+ Map<String, String> additionalProperties = buildAdditionalProperties(event);
+ if (!additionalProperties.isEmpty()) {
+ polarisEvent.setAdditionalProperties(additionalProperties);
+ }
+
processEvent(event.metadata().realmId(), polarisEvent);
}
- private void handleAfterCreateCatalog(PolarisEvent event) {
- Catalog catalog = event.attributes().getRequired(EventAttributes.CATALOG);
- org.apache.polaris.core.entity.PolarisEvent polarisEvent =
- new org.apache.polaris.core.entity.PolarisEvent(
- catalog.getName(),
- event.metadata().eventId().toString(),
- event.metadata().requestId().orElse(null),
- event.type().name(),
- event.metadata().timestamp().toEpochMilli(),
- event.metadata().user().map(PolarisPrincipal::getName).orElse(null),
- org.apache.polaris.core.entity.PolarisEvent.ResourceType.CATALOG,
- catalog.getName());
- Map<String, String> openTelemetryContext = event.metadata().openTelemetryContext();
- if (!openTelemetryContext.isEmpty()) {
- polarisEvent.setAdditionalProperties(openTelemetryContext);
+ private static String resolveCatalogName(PolarisEvent event) {
+ return event
+ .attributes()
+ .get(EventAttributes.CATALOG_NAME)
+ .orElse(org.apache.polaris.core.entity.PolarisEvent.REALM_SCOPED);
+ }
+
+ /**
+ * Maps an event's category to a {@link ResourceType} for indexing. Exhaustive over all {@link
+ * PolarisEventType.Category} values: any new category added to the enum must be classified here
+ * explicitly. The {@code default} arm is intentionally absent so that adding a category produces
+ * a compile-time error rather than silently routing to {@link ResourceType#REALM}.
+ */
+ private static ResourceType resolveResourceType(PolarisEventType eventType) {
+ return switch (eventType.category()) {
+ case TABLE, GENERIC_TABLE -> ResourceType.TABLE;
+ case VIEW -> ResourceType.VIEW;
+ case NAMESPACE -> ResourceType.NAMESPACE;
+ case CATALOG -> ResourceType.CATALOG;
+ case POLICY,
+ PRINCIPAL,
+ PRINCIPAL_ROLE,
+ CATALOG_ROLE,
+ CREDENTIAL,
+ TRANSACTION,
+ NOTIFICATION,
+ CONFIG,
+ TASK_EXECUTION,
+ RATE_LIMITING ->
+ ResourceType.REALM;
+ };
+ }
+
+ private static String resolveResourceIdentifier(
+ PolarisEvent event, ResourceType resourceType, String catalogName) {
+ return switch (resourceType) {
+ case TABLE -> resolveTableResourceIdentifier(event, catalogName);
+ case VIEW -> resolveViewResourceIdentifier(event, catalogName);
+ case NAMESPACE -> resolveNamespaceResourceIdentifier(event, catalogName);
+ case CATALOG -> resolveCatalogResourceIdentifier(event, catalogName);
+ case REALM -> resolveRealmResourceIdentifier(event);
+ };
+ }
+
+ private static String resolveTableResourceIdentifier(PolarisEvent event, String catalogName) {
Review Comment:
The logic in this method is solid, but strikes me as overly convoluted. It probably hints at something that could be optimized on the event emission side.
For instance, why have both `TABLE_IDENTIFIER` and `TABLE_NAME` attributes? I briefly checked all the call sites that use `TABLE_NAME` and it seems doable to use `TABLE_IDENTIFIER` instead. By using `TABLE_IDENTIFIER` consistently we could eliminate the `resolveNamespaceAndTableName` step below.
For events with a `CREATE_TABLE_REQUEST` or `REGISTER_TABLE_REQUEST` attribute, we could also emit a `TABLE_IDENTIFIER` attribute as well.
Moreover, we could consider that most (all?) events target _some_ polaris entity. Having a global attribute `TARGET_ENTITY_ID` would also eliminate most of the logic here, while also benefitting to other listeners as well (as opposed to any logic done here, which only benefits to the persistence listener). We could leverage `EventSanitizer` for this โ see my other comments.
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -86,12 +88,14 @@ private void deliverEvent(
PolarisEvent event, String listenerName, PolarisEventListener listener) {
LOGGER.debug("Delivering {} event to listener '{}' ({})", event.type(), listenerName, listener);
try {
+ PolarisEvent toDeliver =
+ (listener instanceof RawEventAccess) ? event : eventSanitizer.sanitize(event);
Review Comment:
I agree that sanitizers should execute on the Vert.x event loop, as you did here โ but this comes with an important caveat highlighted by #4648 : sanitizers **MUST NOT** block. Maybe we should add an `@implNote` to the `EventSanitizer` interface?
##########
runtime/service/src/main/java/org/apache/polaris/service/events/listeners/PolarisPersistenceEventListener.java:
##########
@@ -60,34 +55,256 @@ private void handleAfterCreateTable(PolarisEvent event) {
event.type().name(),
event.metadata().timestamp().toEpochMilli(),
event.metadata().user().map(PolarisPrincipal::getName).orElse(null),
- org.apache.polaris.core.entity.PolarisEvent.ResourceType.TABLE,
- TableIdentifier.of(namespace, tableName).toString());
- var additionalParameters =
- ImmutableMap.<String, String>builder()
- .put("table-uuid", tableMetadata.uuid())
- .put("metadata", TableMetadataParser.toJson(tableMetadata));
- additionalParameters.putAll(event.metadata().openTelemetryContext());
- polarisEvent.setAdditionalProperties(additionalParameters.build());
+ resourceType,
+ resourceIdentifier);
+
+ Map<String, String> additionalProperties = buildAdditionalProperties(event);
+ if (!additionalProperties.isEmpty()) {
+ polarisEvent.setAdditionalProperties(additionalProperties);
+ }
+
processEvent(event.metadata().realmId(), polarisEvent);
}
- private void handleAfterCreateCatalog(PolarisEvent event) {
- Catalog catalog = event.attributes().getRequired(EventAttributes.CATALOG);
- org.apache.polaris.core.entity.PolarisEvent polarisEvent =
- new org.apache.polaris.core.entity.PolarisEvent(
- catalog.getName(),
- event.metadata().eventId().toString(),
- event.metadata().requestId().orElse(null),
- event.type().name(),
- event.metadata().timestamp().toEpochMilli(),
- event.metadata().user().map(PolarisPrincipal::getName).orElse(null),
- org.apache.polaris.core.entity.PolarisEvent.ResourceType.CATALOG,
- catalog.getName());
- Map<String, String> openTelemetryContext = event.metadata().openTelemetryContext();
- if (!openTelemetryContext.isEmpty()) {
- polarisEvent.setAdditionalProperties(openTelemetryContext);
+ private static String resolveCatalogName(PolarisEvent event) {
+ return event
+ .attributes()
+ .get(EventAttributes.CATALOG_NAME)
+ .orElse(org.apache.polaris.core.entity.PolarisEvent.REALM_SCOPED);
+ }
+
+ /**
+ * Maps an event's category to a {@link ResourceType} for indexing. Exhaustive over all {@link
+ * PolarisEventType.Category} values: any new category added to the enum must be classified here
+ * explicitly. The {@code default} arm is intentionally absent so that adding a category produces
+ * a compile-time error rather than silently routing to {@link ResourceType#REALM}.
+ */
+ private static ResourceType resolveResourceType(PolarisEventType eventType) {
+ return switch (eventType.category()) {
+ case TABLE, GENERIC_TABLE -> ResourceType.TABLE;
+ case VIEW -> ResourceType.VIEW;
+ case NAMESPACE -> ResourceType.NAMESPACE;
+ case CATALOG -> ResourceType.CATALOG;
+ case POLICY,
+ PRINCIPAL,
+ PRINCIPAL_ROLE,
+ CATALOG_ROLE,
+ CREDENTIAL,
+ TRANSACTION,
+ NOTIFICATION,
+ CONFIG,
+ TASK_EXECUTION,
+ RATE_LIMITING ->
+ ResourceType.REALM;
+ };
+ }
+
+ private static String resolveResourceIdentifier(
+ PolarisEvent event, ResourceType resourceType, String catalogName) {
+ return switch (resourceType) {
+ case TABLE -> resolveTableResourceIdentifier(event, catalogName);
+ case VIEW -> resolveViewResourceIdentifier(event, catalogName);
+ case NAMESPACE -> resolveNamespaceResourceIdentifier(event, catalogName);
+ case CATALOG -> resolveCatalogResourceIdentifier(event, catalogName);
+ case REALM -> resolveRealmResourceIdentifier(event);
+ };
+ }
+
+ private static String resolveTableResourceIdentifier(PolarisEvent event, String catalogName) {
+ EventAttributeMap attributes = event.attributes();
+
+ Optional<String> identifierFromTableAttribute =
+ attributes.get(EventAttributes.TABLE_IDENTIFIER).map(TableIdentifier::toString);
+ if (identifierFromTableAttribute.isPresent()) {
+ return identifierFromTableAttribute.get();
}
- processEvent(event.metadata().realmId(), polarisEvent);
+
+ Optional<String> renameIdentifier = resolveRenameIdentifier(event);
+ if (renameIdentifier.isPresent()) {
+ return renameIdentifier.get();
+ }
+
+ Optional<String> namespaceAndTableName = resolveNamespaceAndTableName(attributes);
+ if (namespaceAndTableName.isPresent()) {
+ return namespaceAndTableName.get();
+ }
+
+ Optional<String> namespaceAndCreateTableName = resolveCreateTableIdentifier(attributes);
+ if (namespaceAndCreateTableName.isPresent()) {
+ return namespaceAndCreateTableName.get();
+ }
+
+ Optional<String> namespaceAndRegisterTableName = resolveRegisterTableIdentifier(attributes);
+ if (namespaceAndRegisterTableName.isPresent()) {
+ return namespaceAndRegisterTableName.get();
+ }
+
+ Optional<String> namespaceIdentifier =
+ attributes.get(EventAttributes.NAMESPACE).map(Namespace::toString);
+ if (namespaceIdentifier.isPresent()) {
+ return namespaceIdentifier.get();
+ }
+
+ return attributes
+ .get(EventAttributes.TABLE_NAME)
+ .orElseGet(() -> fallbackResourceIdentifier(event, catalogName));
+ }
+
+ private static String resolveViewResourceIdentifier(PolarisEvent event, String catalogName) {
+ EventAttributeMap attributes = event.attributes();
+ return attributes
+ .get(EventAttributes.VIEW_IDENTIFIER)
+ .map(TableIdentifier::toString)
+ .or(() -> resolveRenameIdentifier(event))
+ .or(
+ () ->
+ attributes
+ .get(EventAttributes.NAMESPACE)
+ .flatMap(
+ namespace ->
+ attributes
+ .get(EventAttributes.VIEW_NAME)
+ .map(
+ viewName ->
+ TableIdentifier.of(namespace, viewName).toString())))
+ .or(() -> attributes.get(EventAttributes.VIEW_NAME))
+ .orElseGet(() -> fallbackResourceIdentifier(event, catalogName));
+ }
+
+ private static Optional<String> resolveRenameIdentifier(PolarisEvent event) {
+ return event
+ .attributes()
+ .get(EventAttributes.RENAME_TABLE_REQUEST)
+ .map(
+ request ->
+ event.type().name().startsWith("AFTER_")
+ ? request.destination().toString()
+ : request.source().toString());
+ }
+
+ private static Optional<String> resolveNamespaceAndTableName(EventAttributeMap attributes) {
+ return attributes
+ .get(EventAttributes.NAMESPACE)
+ .flatMap(
+ namespace ->
+ attributes
+ .get(EventAttributes.TABLE_NAME)
+ .map(tableName -> TableIdentifier.of(namespace, tableName).toString()));
+ }
+
+ private static Optional<String> resolveCreateTableIdentifier(EventAttributeMap attributes) {
+ return attributes
+ .get(EventAttributes.NAMESPACE)
+ .flatMap(
+ namespace ->
+ attributes
+ .get(EventAttributes.CREATE_TABLE_REQUEST)
+ .map(CreateTableRequest::name)
+ .map(tableName -> TableIdentifier.of(namespace, tableName).toString()));
+ }
+
+ private static Optional<String> resolveRegisterTableIdentifier(EventAttributeMap attributes) {
+ return attributes
+ .get(EventAttributes.NAMESPACE)
+ .flatMap(
+ namespace ->
+ attributes
+ .get(EventAttributes.REGISTER_TABLE_REQUEST)
+ .map(RegisterTableRequest::name)
+ .map(tableName -> TableIdentifier.of(namespace, tableName).toString()));
+ }
+
+ private static String resolveNamespaceResourceIdentifier(PolarisEvent event, String catalogName) {
+ EventAttributeMap attributes = event.attributes();
+ return attributes
+ .get(EventAttributes.NAMESPACE)
+ .map(Namespace::toString)
+ .or(() -> attributes.get(EventAttributes.NAMESPACE_FQN))
+ .or(() -> attributes.get(EventAttributes.PARENT_NAMESPACE_FQN))
+ .orElseGet(() -> fallbackResourceIdentifier(event, catalogName));
+ }
+
+ private static String resolveCatalogResourceIdentifier(PolarisEvent event, String catalogName) {
+ return event
+ .attributes()
+ .get(EventAttributes.CATALOG_NAME)
+ .orElseGet(() -> fallbackResourceIdentifier(event, catalogName));
+ }
+
+ private static String resolveRealmResourceIdentifier(PolarisEvent event) {
+ return event.type().name();
+ }
+
+ private static String fallbackResourceIdentifier(PolarisEvent event, String catalogName) {
+ if (!org.apache.polaris.core.entity.PolarisEvent.REALM_SCOPED.equals(catalogName)) {
+ return catalogName;
+ }
+ return event.type().name();
Review Comment:
That's a strange default value, especially given that the event type is already stored in the `event_type` column. Since `resource_identifier` is also `NOT NULL` we should either 1) insert a sentinel value like you did for `catalog_id` or 2) reject this event as malformed. Wdyt?
##########
runtime/service/src/main/java/org/apache/polaris/service/events/listeners/PolarisPersistenceEventListener.java:
##########
@@ -60,34 +55,256 @@ private void handleAfterCreateTable(PolarisEvent event) {
event.type().name(),
event.metadata().timestamp().toEpochMilli(),
event.metadata().user().map(PolarisPrincipal::getName).orElse(null),
- org.apache.polaris.core.entity.PolarisEvent.ResourceType.TABLE,
- TableIdentifier.of(namespace, tableName).toString());
- var additionalParameters =
- ImmutableMap.<String, String>builder()
- .put("table-uuid", tableMetadata.uuid())
- .put("metadata", TableMetadataParser.toJson(tableMetadata));
- additionalParameters.putAll(event.metadata().openTelemetryContext());
- polarisEvent.setAdditionalProperties(additionalParameters.build());
+ resourceType,
+ resourceIdentifier);
+
+ Map<String, String> additionalProperties = buildAdditionalProperties(event);
+ if (!additionalProperties.isEmpty()) {
+ polarisEvent.setAdditionalProperties(additionalProperties);
+ }
+
processEvent(event.metadata().realmId(), polarisEvent);
}
- private void handleAfterCreateCatalog(PolarisEvent event) {
- Catalog catalog = event.attributes().getRequired(EventAttributes.CATALOG);
- org.apache.polaris.core.entity.PolarisEvent polarisEvent =
- new org.apache.polaris.core.entity.PolarisEvent(
- catalog.getName(),
- event.metadata().eventId().toString(),
- event.metadata().requestId().orElse(null),
- event.type().name(),
- event.metadata().timestamp().toEpochMilli(),
- event.metadata().user().map(PolarisPrincipal::getName).orElse(null),
- org.apache.polaris.core.entity.PolarisEvent.ResourceType.CATALOG,
- catalog.getName());
- Map<String, String> openTelemetryContext = event.metadata().openTelemetryContext();
- if (!openTelemetryContext.isEmpty()) {
- polarisEvent.setAdditionalProperties(openTelemetryContext);
+ private static String resolveCatalogName(PolarisEvent event) {
+ return event
+ .attributes()
+ .get(EventAttributes.CATALOG_NAME)
+ .orElse(org.apache.polaris.core.entity.PolarisEvent.REALM_SCOPED);
+ }
+
+ /**
+ * Maps an event's category to a {@link ResourceType} for indexing. Exhaustive over all {@link
+ * PolarisEventType.Category} values: any new category added to the enum must be classified here
+ * explicitly. The {@code default} arm is intentionally absent so that adding a category produces
+ * a compile-time error rather than silently routing to {@link ResourceType#REALM}.
+ */
+ private static ResourceType resolveResourceType(PolarisEventType eventType) {
+ return switch (eventType.category()) {
+ case TABLE, GENERIC_TABLE -> ResourceType.TABLE;
+ case VIEW -> ResourceType.VIEW;
+ case NAMESPACE -> ResourceType.NAMESPACE;
+ case CATALOG -> ResourceType.CATALOG;
+ case POLICY,
+ PRINCIPAL,
+ PRINCIPAL_ROLE,
+ CATALOG_ROLE,
+ CREDENTIAL,
+ TRANSACTION,
+ NOTIFICATION,
+ CONFIG,
+ TASK_EXECUTION,
+ RATE_LIMITING ->
+ ResourceType.REALM;
+ };
+ }
+
+ private static String resolveResourceIdentifier(
+ PolarisEvent event, ResourceType resourceType, String catalogName) {
+ return switch (resourceType) {
+ case TABLE -> resolveTableResourceIdentifier(event, catalogName);
+ case VIEW -> resolveViewResourceIdentifier(event, catalogName);
+ case NAMESPACE -> resolveNamespaceResourceIdentifier(event, catalogName);
+ case CATALOG -> resolveCatalogResourceIdentifier(event, catalogName);
+ case REALM -> resolveRealmResourceIdentifier(event);
+ };
+ }
+
+ private static String resolveTableResourceIdentifier(PolarisEvent event, String catalogName) {
+ EventAttributeMap attributes = event.attributes();
+
+ Optional<String> identifierFromTableAttribute =
+ attributes.get(EventAttributes.TABLE_IDENTIFIER).map(TableIdentifier::toString);
+ if (identifierFromTableAttribute.isPresent()) {
+ return identifierFromTableAttribute.get();
}
- processEvent(event.metadata().realmId(), polarisEvent);
+
+ Optional<String> renameIdentifier = resolveRenameIdentifier(event);
+ if (renameIdentifier.isPresent()) {
+ return renameIdentifier.get();
+ }
+
+ Optional<String> namespaceAndTableName = resolveNamespaceAndTableName(attributes);
+ if (namespaceAndTableName.isPresent()) {
+ return namespaceAndTableName.get();
+ }
+
+ Optional<String> namespaceAndCreateTableName = resolveCreateTableIdentifier(attributes);
+ if (namespaceAndCreateTableName.isPresent()) {
+ return namespaceAndCreateTableName.get();
+ }
+
+ Optional<String> namespaceAndRegisterTableName = resolveRegisterTableIdentifier(attributes);
+ if (namespaceAndRegisterTableName.isPresent()) {
+ return namespaceAndRegisterTableName.get();
+ }
+
+ Optional<String> namespaceIdentifier =
+ attributes.get(EventAttributes.NAMESPACE).map(Namespace::toString);
+ if (namespaceIdentifier.isPresent()) {
+ return namespaceIdentifier.get();
+ }
+
+ return attributes
+ .get(EventAttributes.TABLE_NAME)
+ .orElseGet(() -> fallbackResourceIdentifier(event, catalogName));
+ }
+
+ private static String resolveViewResourceIdentifier(PolarisEvent event, String catalogName) {
+ EventAttributeMap attributes = event.attributes();
+ return attributes
+ .get(EventAttributes.VIEW_IDENTIFIER)
+ .map(TableIdentifier::toString)
+ .or(() -> resolveRenameIdentifier(event))
+ .or(
+ () ->
+ attributes
+ .get(EventAttributes.NAMESPACE)
+ .flatMap(
+ namespace ->
+ attributes
+ .get(EventAttributes.VIEW_NAME)
+ .map(
+ viewName ->
+ TableIdentifier.of(namespace, viewName).toString())))
+ .or(() -> attributes.get(EventAttributes.VIEW_NAME))
+ .orElseGet(() -> fallbackResourceIdentifier(event, catalogName));
+ }
+
+ private static Optional<String> resolveRenameIdentifier(PolarisEvent event) {
+ return event
+ .attributes()
+ .get(EventAttributes.RENAME_TABLE_REQUEST)
+ .map(
+ request ->
+ event.type().name().startsWith("AFTER_")
+ ? request.destination().toString()
+ : request.source().toString());
+ }
+
+ private static Optional<String> resolveNamespaceAndTableName(EventAttributeMap attributes) {
+ return attributes
+ .get(EventAttributes.NAMESPACE)
+ .flatMap(
+ namespace ->
+ attributes
+ .get(EventAttributes.TABLE_NAME)
+ .map(tableName -> TableIdentifier.of(namespace, tableName).toString()));
+ }
+
+ private static Optional<String> resolveCreateTableIdentifier(EventAttributeMap attributes) {
+ return attributes
+ .get(EventAttributes.NAMESPACE)
+ .flatMap(
+ namespace ->
+ attributes
+ .get(EventAttributes.CREATE_TABLE_REQUEST)
+ .map(CreateTableRequest::name)
+ .map(tableName -> TableIdentifier.of(namespace, tableName).toString()));
+ }
+
+ private static Optional<String> resolveRegisterTableIdentifier(EventAttributeMap attributes) {
+ return attributes
+ .get(EventAttributes.NAMESPACE)
+ .flatMap(
+ namespace ->
+ attributes
+ .get(EventAttributes.REGISTER_TABLE_REQUEST)
+ .map(RegisterTableRequest::name)
+ .map(tableName -> TableIdentifier.of(namespace, tableName).toString()));
+ }
+
+ private static String resolveNamespaceResourceIdentifier(PolarisEvent event, String catalogName) {
+ EventAttributeMap attributes = event.attributes();
+ return attributes
+ .get(EventAttributes.NAMESPACE)
+ .map(Namespace::toString)
+ .or(() -> attributes.get(EventAttributes.NAMESPACE_FQN))
+ .or(() -> attributes.get(EventAttributes.PARENT_NAMESPACE_FQN))
+ .orElseGet(() -> fallbackResourceIdentifier(event, catalogName));
+ }
+
+ private static String resolveCatalogResourceIdentifier(PolarisEvent event, String catalogName) {
+ return event
+ .attributes()
+ .get(EventAttributes.CATALOG_NAME)
+ .orElseGet(() -> fallbackResourceIdentifier(event, catalogName));
+ }
+
+ private static String resolveRealmResourceIdentifier(PolarisEvent event) {
+ return event.type().name();
+ }
+
+ private static String fallbackResourceIdentifier(PolarisEvent event, String catalogName) {
+ if (!org.apache.polaris.core.entity.PolarisEvent.REALM_SCOPED.equals(catalogName)) {
+ return catalogName;
+ }
+ return event.type().name();
+ }
+
+ private static Map<String, String> buildAdditionalProperties(PolarisEvent event) {
+ Map<String, String> additionalProperties =
+ new LinkedHashMap<>(event.metadata().openTelemetryContext());
+ event.attributes().forEach((key, value) -> additionalProperties.putAll(prune(key, value)));
+ return additionalProperties;
+ }
+
+ /**
+ * Reduces an attribute value to a bounded string representation suitable for storage in the
+ * persistence event row. Inlined here (rather than exposed as a swappable bean) because pruning
+ * is a serialization concern of this listener only โ other listeners are free to render the same
+ * value differently.
+ */
+ private static Map<String, String> prune(AttributeKey<?> key, Object value) {
+ if (value instanceof String || value instanceof Number || value instanceof Boolean) {
+ return Map.of(key.name(), value.toString());
+ }
+
+ if (value instanceof Namespace namespace) {
+ return Map.of(key.name(), namespace.toString());
+ }
+
+ if (value instanceof TableIdentifier tableIdentifier) {
+ return Map.of(key.name(), tableIdentifier.toString());
+ }
+
+ if (key.equals(EventAttributes.TABLE_METADATA) && value instanceof TableMetadata metadata) {
+ return pruneTableMetadata(metadata);
+ }
+
+ if (key.equals(EventAttributes.LOAD_TABLE_RESPONSE)
+ && value instanceof LoadTableResponse response) {
+ return pruneTableMetadata(response.tableMetadata());
+ }
+
+ if (key.equals(EventAttributes.RENAME_TABLE_REQUEST)
+ && value instanceof RenameTableRequest request) {
+ Map<String, String> result = new LinkedHashMap<>();
+ result.put("rename_source", request.source().toString());
Review Comment:
I wonder if these shouldn't be standard attributes emitted on every rename operation? Every listener dealing with a rename-table event will have to compute these values.
##########
runtime/service/src/main/java/org/apache/polaris/service/events/DefaultEventSanitizer.java:
##########
@@ -0,0 +1,127 @@
+/*
+ * 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 io.quarkus.arc.DefaultBean;
+import jakarta.annotation.PostConstruct;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Set;
+import org.apache.polaris.core.admin.model.Catalog;
+
+/**
+ * Default implementation of {@link EventSanitizer}. Drops attributes whose keys are on the denylist
+ * and derives safe attributes (e.g., {@code CATALOG_NAME} extracted from a denied {@code CATALOG}
+ * object) from the originals. The built-in denylist covers attributes that carry credentials or
+ * internal secrets and may be extended via {@code polaris.event-listener.denylisted-attributes}.
+ *
+ * <p>Operators who need a different denial or derivation policy should replace this bean.
+ */
+@ApplicationScoped
+@DefaultBean
+public class DefaultEventSanitizer implements EventSanitizer {
+
+ // These carry full domain objects containing credentials or internal secrets:
+ // PRINCIPAL - may contain client IDs, internal identity metadata
+ // UPDATE_PRINCIPAL_REQUEST - credential rotation data, secret changes
+ // CATALOG - includes StorageConfigInfo with cloud credentials (AWS keys, Azure tokens, etc.)
+ // NOTIFICATION_REQUEST - contains metadata file locations and potentially signed URLs
+ static final Set<AttributeKey<?>> DEFAULT_DENYLIST =
+ Set.of(
+ EventAttributes.PRINCIPAL,
+ EventAttributes.UPDATE_PRINCIPAL_REQUEST,
+ EventAttributes.CATALOG,
+ EventAttributes.NOTIFICATION_REQUEST);
+
+ @Inject PolarisEventListenerConfiguration configuration;
+
+ private Set<AttributeKey<?>> denylist;
+
+ DefaultEventSanitizer() {
+ this(DEFAULT_DENYLIST);
+ }
+
+ DefaultEventSanitizer(Set<AttributeKey<?>> denylist) {
+ this.denylist = Set.copyOf(denylist);
+ }
+
+ @PostConstruct
+ void initializeFromConfiguration() {
+ if (configuration == null) {
+ return;
+ }
+ Set<String> configuredNames =
+ configuration.denylistedAttributes().orElse(Collections.emptySet());
+ if (!configuredNames.isEmpty()) {
+ this.denylist = resolveAdditionalDenylist(configuredNames);
+ }
+ }
+
+ @Override
+ public PolarisEvent sanitize(PolarisEvent event) {
+ EventAttributeMap filtered = new EventAttributeMap();
+ event
+ .attributes()
+ .forEach(
+ (key, value) -> {
+ if (!denylist.contains(key)) {
+ putUnchecked(filtered, key, value);
+ }
+ });
+ extractDerivedAttributes(event, filtered);
+ return new PolarisEvent(event.type(), event.metadata(), filtered);
+ }
+
+ public static Set<AttributeKey<?>> resolveAdditionalDenylist(Set<String> configuredNames) {
+ Set<AttributeKey<?>> resolved = new LinkedHashSet<>(DEFAULT_DENYLIST);
+ List<String> unknown = new ArrayList<>();
+
+ for (String name : configuredNames) {
+ EventAttributes.findByName(name).ifPresentOrElse(resolved::add, () -> unknown.add(name));
+ }
+
+ if (!unknown.isEmpty()) {
+ throw new IllegalArgumentException(
+ "Unknown event attributes in denylist configuration: " + unknown);
+ }
+
+ return Set.copyOf(resolved);
+ }
+
+ private static void extractDerivedAttributes(PolarisEvent event, EventAttributeMap filtered) {
Review Comment:
Oh I just looked at the `shouldDeriveCatalogNameFromCatalogObjectDuringFiltering` test and I think I get it. But it's too rigid imho. We could do the same for table or view metadata, etc.
Imho a better approach is the one I outlined in another comment: a sanitizer would first enrich the event with derived attributes (e.g. catalog ID from catalog), then a second sanitizer would remove the catalog attribute.
##########
runtime/service/src/main/java/org/apache/polaris/service/events/DefaultEventSanitizer.java:
##########
@@ -0,0 +1,127 @@
+/*
+ * 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 io.quarkus.arc.DefaultBean;
+import jakarta.annotation.PostConstruct;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Set;
+import org.apache.polaris.core.admin.model.Catalog;
+
+/**
+ * Default implementation of {@link EventSanitizer}. Drops attributes whose keys are on the denylist
+ * and derives safe attributes (e.g., {@code CATALOG_NAME} extracted from a denied {@code CATALOG}
+ * object) from the originals. The built-in denylist covers attributes that carry credentials or
+ * internal secrets and may be extended via {@code polaris.event-listener.denylisted-attributes}.
+ *
+ * <p>Operators who need a different denial or derivation policy should replace this bean.
Review Comment:
That's a bit rigid, as I stated above. I think we could achieve the same without requiring users to modify Polaris and rebuild it.
##########
runtime/service/src/main/java/org/apache/polaris/service/events/DefaultEventSanitizer.java:
##########
@@ -0,0 +1,127 @@
+/*
+ * 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 io.quarkus.arc.DefaultBean;
+import jakarta.annotation.PostConstruct;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Set;
+import org.apache.polaris.core.admin.model.Catalog;
+
+/**
+ * Default implementation of {@link EventSanitizer}. Drops attributes whose keys are on the denylist
+ * and derives safe attributes (e.g., {@code CATALOG_NAME} extracted from a denied {@code CATALOG}
+ * object) from the originals. The built-in denylist covers attributes that carry credentials or
+ * internal secrets and may be extended via {@code polaris.event-listener.denylisted-attributes}.
+ *
+ * <p>Operators who need a different denial or derivation policy should replace this bean.
+ */
+@ApplicationScoped
+@DefaultBean
+public class DefaultEventSanitizer implements EventSanitizer {
+
+ // These carry full domain objects containing credentials or internal secrets:
+ // PRINCIPAL - may contain client IDs, internal identity metadata
+ // UPDATE_PRINCIPAL_REQUEST - credential rotation data, secret changes
+ // CATALOG - includes StorageConfigInfo with cloud credentials (AWS keys, Azure tokens, etc.)
+ // NOTIFICATION_REQUEST - contains metadata file locations and potentially signed URLs
+ static final Set<AttributeKey<?>> DEFAULT_DENYLIST =
+ Set.of(
+ EventAttributes.PRINCIPAL,
+ EventAttributes.UPDATE_PRINCIPAL_REQUEST,
+ EventAttributes.CATALOG,
+ EventAttributes.NOTIFICATION_REQUEST);
+
+ @Inject PolarisEventListenerConfiguration configuration;
+
+ private Set<AttributeKey<?>> denylist;
+
+ DefaultEventSanitizer() {
+ this(DEFAULT_DENYLIST);
+ }
+
+ DefaultEventSanitizer(Set<AttributeKey<?>> denylist) {
+ this.denylist = Set.copyOf(denylist);
+ }
+
+ @PostConstruct
+ void initializeFromConfiguration() {
+ if (configuration == null) {
+ return;
+ }
+ Set<String> configuredNames =
+ configuration.denylistedAttributes().orElse(Collections.emptySet());
+ if (!configuredNames.isEmpty()) {
+ this.denylist = resolveAdditionalDenylist(configuredNames);
+ }
+ }
+
+ @Override
+ public PolarisEvent sanitize(PolarisEvent event) {
+ EventAttributeMap filtered = new EventAttributeMap();
+ event
+ .attributes()
+ .forEach(
+ (key, value) -> {
+ if (!denylist.contains(key)) {
+ putUnchecked(filtered, key, value);
+ }
+ });
+ extractDerivedAttributes(event, filtered);
+ return new PolarisEvent(event.type(), event.metadata(), filtered);
+ }
+
+ public static Set<AttributeKey<?>> resolveAdditionalDenylist(Set<String> configuredNames) {
+ Set<AttributeKey<?>> resolved = new LinkedHashSet<>(DEFAULT_DENYLIST);
+ List<String> unknown = new ArrayList<>();
+
+ for (String name : configuredNames) {
+ EventAttributes.findByName(name).ifPresentOrElse(resolved::add, () -> unknown.add(name));
+ }
+
+ if (!unknown.isEmpty()) {
+ throw new IllegalArgumentException(
+ "Unknown event attributes in denylist configuration: " + unknown);
+ }
+
+ return Set.copyOf(resolved);
+ }
+
+ private static void extractDerivedAttributes(PolarisEvent event, EventAttributeMap filtered) {
+ if (!filtered.contains(EventAttributes.CATALOG_NAME)) {
+ event
+ .attributes()
+ .get(EventAttributes.CATALOG)
+ .map(Catalog::getName)
+ .filter(name -> !name.isBlank())
+ .ifPresent(name -> filtered.put(EventAttributes.CATALOG_NAME, name));
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <T> void putUnchecked(EventAttributeMap map, AttributeKey<T> key, Object value) {
+ map.put(key, (T) value);
+ }
Review Comment:
nit: Slightly better:
```suggestion
private static <T> void putUnchecked(EventAttributeMap map, AttributeKey<T> key, Object value) {
map.put(key, key.cast(value));
}
```
(But you still need the method โ you can't inline it.)
##########
runtime/service/src/main/java/org/apache/polaris/service/events/EventAttributes.java:
##########
@@ -240,4 +247,42 @@ private EventAttributes() {}
new AttributeKey<>("detach_policy_request", DetachPolicyRequest.class);
public static final AttributeKey<GetApplicablePoliciesResponse> GET_APPLICABLE_POLICIES_RESPONSE =
new AttributeKey<>("get_applicable_policies_response", GetApplicablePoliciesResponse.class);
+
+ public static Optional<AttributeKey<?>> findByName(String name) {
+ return Optional.ofNullable(AttributeLookupHolder.ALL_BY_NAME.get(name));
+ }
+
+ public static Map<String, AttributeKey<?>> allByName() {
+ return AttributeLookupHolder.ALL_BY_NAME;
+ }
+
+ private static Map<String, AttributeKey<?>> collectAllByName() {
Review Comment:
Nice addition, and the lazy init is ๐
##########
runtime/service/src/main/java/org/apache/polaris/service/events/DefaultEventSanitizer.java:
##########
@@ -0,0 +1,127 @@
+/*
+ * 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 io.quarkus.arc.DefaultBean;
+import jakarta.annotation.PostConstruct;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Set;
+import org.apache.polaris.core.admin.model.Catalog;
+
+/**
+ * Default implementation of {@link EventSanitizer}. Drops attributes whose keys are on the denylist
+ * and derives safe attributes (e.g., {@code CATALOG_NAME} extracted from a denied {@code CATALOG}
+ * object) from the originals. The built-in denylist covers attributes that carry credentials or
+ * internal secrets and may be extended via {@code polaris.event-listener.denylisted-attributes}.
+ *
+ * <p>Operators who need a different denial or derivation policy should replace this bean.
+ */
+@ApplicationScoped
+@DefaultBean
+public class DefaultEventSanitizer implements EventSanitizer {
+
+ // These carry full domain objects containing credentials or internal secrets:
+ // PRINCIPAL - may contain client IDs, internal identity metadata
+ // UPDATE_PRINCIPAL_REQUEST - credential rotation data, secret changes
+ // CATALOG - includes StorageConfigInfo with cloud credentials (AWS keys, Azure tokens, etc.)
+ // NOTIFICATION_REQUEST - contains metadata file locations and potentially signed URLs
+ static final Set<AttributeKey<?>> DEFAULT_DENYLIST =
+ Set.of(
+ EventAttributes.PRINCIPAL,
+ EventAttributes.UPDATE_PRINCIPAL_REQUEST,
+ EventAttributes.CATALOG,
+ EventAttributes.NOTIFICATION_REQUEST);
+
+ @Inject PolarisEventListenerConfiguration configuration;
+
+ private Set<AttributeKey<?>> denylist;
+
+ DefaultEventSanitizer() {
+ this(DEFAULT_DENYLIST);
+ }
+
+ DefaultEventSanitizer(Set<AttributeKey<?>> denylist) {
+ this.denylist = Set.copyOf(denylist);
+ }
+
+ @PostConstruct
+ void initializeFromConfiguration() {
+ if (configuration == null) {
+ return;
+ }
+ Set<String> configuredNames =
+ configuration.denylistedAttributes().orElse(Collections.emptySet());
+ if (!configuredNames.isEmpty()) {
+ this.denylist = resolveAdditionalDenylist(configuredNames);
+ }
+ }
+
+ @Override
+ public PolarisEvent sanitize(PolarisEvent event) {
+ EventAttributeMap filtered = new EventAttributeMap();
+ event
+ .attributes()
+ .forEach(
+ (key, value) -> {
+ if (!denylist.contains(key)) {
+ putUnchecked(filtered, key, value);
+ }
+ });
+ extractDerivedAttributes(event, filtered);
+ return new PolarisEvent(event.type(), event.metadata(), filtered);
+ }
+
+ public static Set<AttributeKey<?>> resolveAdditionalDenylist(Set<String> configuredNames) {
+ Set<AttributeKey<?>> resolved = new LinkedHashSet<>(DEFAULT_DENYLIST);
+ List<String> unknown = new ArrayList<>();
+
+ for (String name : configuredNames) {
+ EventAttributes.findByName(name).ifPresentOrElse(resolved::add, () -> unknown.add(name));
+ }
+
+ if (!unknown.isEmpty()) {
+ throw new IllegalArgumentException(
+ "Unknown event attributes in denylist configuration: " + unknown);
+ }
+
+ return Set.copyOf(resolved);
+ }
+
+ private static void extractDerivedAttributes(PolarisEvent event, EventAttributeMap filtered) {
Review Comment:
This looks odd, what is the purpose of this method?
##########
runtime/service/src/main/java/org/apache/polaris/service/events/DefaultEventSanitizer.java:
##########
@@ -0,0 +1,127 @@
+/*
+ * 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 io.quarkus.arc.DefaultBean;
+import jakarta.annotation.PostConstruct;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Set;
+import org.apache.polaris.core.admin.model.Catalog;
+
+/**
+ * Default implementation of {@link EventSanitizer}. Drops attributes whose keys are on the denylist
+ * and derives safe attributes (e.g., {@code CATALOG_NAME} extracted from a denied {@code CATALOG}
+ * object) from the originals. The built-in denylist covers attributes that carry credentials or
+ * internal secrets and may be extended via {@code polaris.event-listener.denylisted-attributes}.
+ *
+ * <p>Operators who need a different denial or derivation policy should replace this bean.
+ */
+@ApplicationScoped
+@DefaultBean
+public class DefaultEventSanitizer implements EventSanitizer {
+
+ // These carry full domain objects containing credentials or internal secrets:
+ // PRINCIPAL - may contain client IDs, internal identity metadata
+ // UPDATE_PRINCIPAL_REQUEST - credential rotation data, secret changes
+ // CATALOG - includes StorageConfigInfo with cloud credentials (AWS keys, Azure tokens, etc.)
+ // NOTIFICATION_REQUEST - contains metadata file locations and potentially signed URLs
+ static final Set<AttributeKey<?>> DEFAULT_DENYLIST =
+ Set.of(
+ EventAttributes.PRINCIPAL,
+ EventAttributes.UPDATE_PRINCIPAL_REQUEST,
+ EventAttributes.CATALOG,
+ EventAttributes.NOTIFICATION_REQUEST);
+
+ @Inject PolarisEventListenerConfiguration configuration;
+
+ private Set<AttributeKey<?>> denylist;
+
+ DefaultEventSanitizer() {
+ this(DEFAULT_DENYLIST);
+ }
+
+ DefaultEventSanitizer(Set<AttributeKey<?>> denylist) {
+ this.denylist = Set.copyOf(denylist);
+ }
+
+ @PostConstruct
+ void initializeFromConfiguration() {
+ if (configuration == null) {
+ return;
+ }
+ Set<String> configuredNames =
+ configuration.denylistedAttributes().orElse(Collections.emptySet());
+ if (!configuredNames.isEmpty()) {
+ this.denylist = resolveAdditionalDenylist(configuredNames);
+ }
+ }
+
+ @Override
+ public PolarisEvent sanitize(PolarisEvent event) {
+ EventAttributeMap filtered = new EventAttributeMap();
+ event
+ .attributes()
+ .forEach(
+ (key, value) -> {
+ if (!denylist.contains(key)) {
+ putUnchecked(filtered, key, value);
+ }
+ });
+ extractDerivedAttributes(event, filtered);
+ return new PolarisEvent(event.type(), event.metadata(), filtered);
+ }
+
+ public static Set<AttributeKey<?>> resolveAdditionalDenylist(Set<String> configuredNames) {
+ Set<AttributeKey<?>> resolved = new LinkedHashSet<>(DEFAULT_DENYLIST);
Review Comment:
IIUC, the default denylist is always applied, regardless of user configuration. I wonder if this isn't too rigid, and also counter-intuitive? What if a super-secure, auditing listener wants to inspect `EventAttributes.PRINCIPAL`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor persistence event listener to a universal registry pattern with strict allowlist serialization [polaris]
Posted by "adutra (via GitHub)" <gi...@apache.org>.
adutra commented on code in PR #4225:
URL: https://github.com/apache/polaris/pull/4225#discussion_r3126350980
##########
runtime/service/src/main/java/org/apache/polaris/service/events/listeners/PersistenceAttributeSecurityFilter.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.listeners;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Set;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableMetadataParser;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.rest.responses.LoadTableResponse;
+import org.apache.polaris.service.events.AttributeKey;
+import org.apache.polaris.service.events.EventAttributeMap;
+import org.apache.polaris.service.events.EventAttributes;
+
+final class PersistenceAttributeSecurityFilter {
+
+ private static final Set<String> ALLOWED_TELEMETRY_CONTEXT_KEYS =
+ Set.of("otel.trace_id", "otel.span_id", "otel.trace_flags", "otel.sampled");
+
+ Map<String, String> filterAllowlistedAttributes(EventAttributeMap attributes) {
+ Map<String, String> safeAttributes = new LinkedHashMap<>();
+ attributes.forEach((key, value) -> addIfAllowlisted(safeAttributes, key, value));
+ return safeAttributes;
+ }
+
+ Map<String, String> filterAllowlistedTelemetryContext(Map<String, String> telemetryContext) {
+ Map<String, String> safeContext = new LinkedHashMap<>();
+ telemetryContext.forEach(
+ (key, value) -> {
+ if (ALLOWED_TELEMETRY_CONTEXT_KEYS.contains(key)) {
+ safeContext.put(key, value);
+ }
+ });
+ return safeContext;
+ }
+
+ @SuppressWarnings("unchecked")
+ private static void addIfAllowlisted(
+ Map<String, String> safeAttributes, AttributeKey<?> key, Object value) {
+ if (EventAttributes.CATALOG_NAME.equals(key)
+ || EventAttributes.TABLE_NAME.equals(key)
+ || EventAttributes.VIEW_NAME.equals(key)
+ || EventAttributes.NAMESPACE_FQN.equals(key)
+ || EventAttributes.PARENT_NAMESPACE_FQN.equals(key)) {
+ safeAttributes.put(key.name(), (String) value);
+ return;
+ }
+
+ if (EventAttributes.NAMESPACE.equals(key)) {
+ safeAttributes.put(key.name(), ((Namespace) value).toString());
+ return;
+ }
+
+ if (EventAttributes.TABLE_IDENTIFIER.equals(key)
+ || EventAttributes.VIEW_IDENTIFIER.equals(key)) {
+ safeAttributes.put(key.name(), ((TableIdentifier) value).toString());
+ return;
+ }
+
+ // TODO: Filter sensitive events and credential-bearing table properties before serializing
+ // table metadata. This and other sensitive event filtering cases will be addressed separately.
+ if (EventAttributes.TABLE_METADATA.equals(key)) {
+ addTableMetadata(safeAttributes, (TableMetadata) value);
+ return;
+ }
+
+ if (EventAttributes.LOAD_TABLE_RESPONSE.equals(key)) {
+ addTableMetadata(safeAttributes, ((LoadTableResponse) value).tableMetadata());
+ return;
+ }
+ }
+
+ private static void addTableMetadata(
+ Map<String, String> safeAttributes, TableMetadata tableMetadata) {
+ if (tableMetadata == null) {
+ return;
+ }
+
+ safeAttributes.put(
+ EventAttributes.TABLE_METADATA.name(), TableMetadataParser.toJson(tableMetadata));
Review Comment:
This may result in hundreds of megabytes of JSON. We must introduce a way to prune it.
##########
runtime/service/src/main/java/org/apache/polaris/service/events/listeners/PersistenceEventHandler.java:
##########
@@ -0,0 +1,27 @@
+/*
+ * 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.listeners;
+
+import org.apache.polaris.service.events.PolarisEvent;
+
+interface PersistenceEventHandler {
Review Comment:
If I understand the intent correctly, this new component would be responsible for 2 things:
* Map a `PolarisEvent` to a Polaris event entity;
* Apply some security-related filtering through the subcomponent `PersistenceAttributeSecurityFilter`.
While the event -> entity mapping makes sense for me, the security-related filter seems to be a materialization of the "transformation layer" that we've been discussing for a while, that would sanitize events to remove things like huge payloads or sensitive data.
If that assumption is correct, I think that this isn't the right layer where the filter should apply. It should imho apply to _all events_ in the pipeline, and _before_ they are handed off to a specific listener.
##########
runtime/service/src/main/java/org/apache/polaris/service/events/listeners/PolarisPersistenceEventListener.java:
##########
@@ -19,75 +19,31 @@
package org.apache.polaris.service.events.listeners;
-import com.google.common.collect.ImmutableMap;
import java.util.Map;
-import org.apache.iceberg.TableMetadata;
-import org.apache.iceberg.TableMetadataParser;
-import org.apache.iceberg.catalog.Namespace;
-import org.apache.iceberg.catalog.TableIdentifier;
-import org.apache.iceberg.rest.responses.LoadTableResponse;
-import org.apache.polaris.core.admin.model.Catalog;
-import org.apache.polaris.core.auth.PolarisPrincipal;
-import org.apache.polaris.service.events.EventAttributes;
+import java.util.Objects;
import org.apache.polaris.service.events.PolarisEvent;
+import org.apache.polaris.service.events.PolarisEventType;
public abstract class PolarisPersistenceEventListener implements PolarisEventListener {
- @Override
- public void onEvent(PolarisEvent event) {
- switch (event.type()) {
- case AFTER_CREATE_TABLE -> handleAfterCreateTable(event);
- case AFTER_CREATE_CATALOG -> handleAfterCreateCatalog(event);
- default -> {
- // Other events not handled by this listener
- }
- }
- }
+ private final PersistenceEventHandler defaultHandler;
+ private final Map<PolarisEventType, PersistenceEventHandler> eventTypeHandlers;
- private void handleAfterCreateTable(PolarisEvent event) {
- LoadTableResponse loadTableResponse =
- event.attributes().getRequired(EventAttributes.LOAD_TABLE_RESPONSE);
- TableMetadata tableMetadata = loadTableResponse.tableMetadata();
- String catalogName = event.attributes().getRequired(EventAttributes.CATALOG_NAME);
- Namespace namespace = event.attributes().getRequired(EventAttributes.NAMESPACE);
- String tableName = event.attributes().getRequired(EventAttributes.TABLE_NAME);
+ protected PolarisPersistenceEventListener() {
Review Comment:
Shouldn't this be handled with CDI? How do you plan to make it configurable for users to pick a special handler for a given event type?
##########
runtime/service/src/main/java/org/apache/polaris/service/events/listeners/DefaultPersistenceEventHandler.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * 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.listeners;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Objects;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.rest.requests.RenameTableRequest;
+import org.apache.polaris.core.admin.model.Catalog;
+import org.apache.polaris.core.admin.model.Principal;
+import org.apache.polaris.core.auth.PolarisPrincipal;
+import org.apache.polaris.core.entity.PolarisEvent.ResourceType;
+import org.apache.polaris.service.events.EventAttributeMap;
+import org.apache.polaris.service.events.EventAttributes;
+import org.apache.polaris.service.events.PolarisEvent;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+final class DefaultPersistenceEventHandler implements PersistenceEventHandler {
+
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(DefaultPersistenceEventHandler.class);
+ private static final String SERIALIZATION_ERROR_PAYLOAD = "{\"error\":\"Serialization failed\"}";
+
+ static final String ALL_CATALOGS_SCOPE = "_all_catalogs";
+ static final String ROOT_NAMESPACE_SCOPE = "_root_namespace";
+ static final String TABLE_SCOPE = "_table_scope";
+ static final String VIEW_SCOPE = "_view_scope";
+
+ private final PersistenceAttributeSecurityFilter securityFilter;
+ private final ObjectMapper objectMapper;
+
+ DefaultPersistenceEventHandler() {
+ this(new PersistenceAttributeSecurityFilter(), new ObjectMapper());
Review Comment:
Both arguments should be injected via CDI. The object mapper to use should be the default one.
##########
runtime/service/src/main/java/org/apache/polaris/service/events/listeners/PersistenceAttributeSecurityFilter.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.listeners;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Set;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableMetadataParser;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.rest.responses.LoadTableResponse;
+import org.apache.polaris.service.events.AttributeKey;
+import org.apache.polaris.service.events.EventAttributeMap;
+import org.apache.polaris.service.events.EventAttributes;
+
+final class PersistenceAttributeSecurityFilter {
Review Comment:
This should probably be made a CDI bean and should be configurable.
Moreover, filtering is not enough: we also need a component that prunes/sanitizes huge payloads, and especially, `TableMetadata` objects.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Overhaul event persistence with universal routing and global secure-by-default sanitization [polaris]
Posted by "flyrain (via GitHub)" <gi...@apache.org>.
flyrain commented on PR #4225:
URL: https://github.com/apache/polaris/pull/4225#issuecomment-4633481034
The PR looks good to me. I will merge it by EOD if there is no further comments.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Overhaul event persistence with universal routing and global secure-by-default sanitization [polaris]
Posted by "sririshindra (via GitHub)" <gi...@apache.org>.
sririshindra commented on code in PR #4225:
URL: https://github.com/apache/polaris/pull/4225#discussion_r3359327421
##########
runtime/service/src/main/java/org/apache/polaris/service/events/listeners/PolarisPersistenceEventListener.java:
##########
@@ -19,38 +19,37 @@
package org.apache.polaris.service.events.listeners;
-import com.google.common.collect.ImmutableMap;
+import jakarta.inject.Inject;
+import java.util.LinkedHashMap;
import java.util.Map;
-import org.apache.iceberg.TableMetadata;
-import org.apache.iceberg.TableMetadataParser;
+import java.util.Optional;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
-import org.apache.iceberg.rest.responses.LoadTableResponse;
-import org.apache.polaris.core.admin.model.Catalog;
+import org.apache.iceberg.rest.requests.CreateTableRequest;
+import org.apache.iceberg.rest.requests.RegisterTableRequest;
import org.apache.polaris.core.auth.PolarisPrincipal;
+import org.apache.polaris.service.events.EventAttributeMap;
import org.apache.polaris.service.events.EventAttributes;
+import org.apache.polaris.service.events.EventPayloadPruner;
import org.apache.polaris.service.events.PolarisEvent;
+import org.apache.polaris.service.events.PolarisEventType;
public abstract class PolarisPersistenceEventListener implements PolarisEventListener {
- @Override
- public void onEvent(PolarisEvent event) {
- switch (event.type()) {
- case AFTER_CREATE_TABLE -> handleAfterCreateTable(event);
- case AFTER_CREATE_CATALOG -> handleAfterCreateCatalog(event);
- default -> {
- // Other events not handled by this listener
- }
- }
+ @Inject EventPayloadPruner payloadPruner;
+
+ protected PolarisPersistenceEventListener() {}
+
+ PolarisPersistenceEventListener(EventPayloadPruner payloadPruner) {
+ this.payloadPruner = payloadPruner;
}
- private void handleAfterCreateTable(PolarisEvent event) {
- LoadTableResponse loadTableResponse =
- event.attributes().getRequired(EventAttributes.LOAD_TABLE_RESPONSE);
- TableMetadata tableMetadata = loadTableResponse.tableMetadata();
- String catalogName = event.attributes().getRequired(EventAttributes.CATALOG_NAME);
- Namespace namespace = event.attributes().getRequired(EventAttributes.NAMESPACE);
- String tableName = event.attributes().getRequired(EventAttributes.TABLE_NAME);
+ @Override
+ public void onEvent(PolarisEvent event) {
+ String catalogName = resolveCatalogName(event);
Review Comment:
You are right that `catalogId` is `NOT NULL` in the persistence schema, so
realm-scoped events (principal, principal-role, policy, etc.) can't store
`NULL` directly today. The `__realm__` sentinel is a transitional placeholder.
Making the column nullable is a schema migration that touches `PolarisEvent`
in `polaris-core` and the underlying persistence โ bigger blast radius than
this PR should carry. I've added an inline `TODO` on
`PolarisEvent.REALM_SCOPED` calling out the planned removal:
```java
// TODO: Remove this sentinel once the catalogId column is made nullable.
// Realm-scoped events should store NULL, not a magic string, so
// downstream queries can use IS NULL instead of filtering out a
// reserved value.
```
I'll file a follow-up issue with the migration plan and link it here once
opened. Happy to keep this PR focused on the routing/sanitization changes
and tackle the schema change separately โ let me know if you'd prefer it
bundled.
##########
runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java:
##########
@@ -80,8 +82,13 @@ public void onStartup(@Observes StartupEvent event) {
private void deliverEvent(
PolarisEvent event, String listenerName, PolarisEventListener listener) {
LOGGER.debug("Delivering {} event to listener '{}' ({})", event.type(), listenerName, listener);
+ // TODO: Add an opt-in mechanism (e.g., a marker interface or annotation) for specialized
Review Comment:
Done โ implemented exactly as suggested:
- Added `org.apache.polaris.service.events.listeners.RawEventAccess`, an empty
marker interface with Javadoc spelling out that it is an explicit
security opt-out and should be implemented only by listeners with a
documented need for sensitive attributes (e.g., forensic audit sinks).
- Updated the dispatcher in `PolarisEventListeners.deliverEvent`:
```java
PolarisEvent toDeliver =
(listener instanceof RawEventAccess) ? event : eventSanitizer.sanitize(event);
listener.onEvent(toDeliver);
```
- Removed the placeholder `TODO` referencing the dev@ thread.
- Added `RawEventAccessTest` (a separate `@QuarkusTest` with its own profile)
to verify a `RawEventAccess` listener receives the unsanitized event with
the `CATALOG` attribute intact.
To answer the standing question: this PR does not introduce per-listener
sanitizer selection. There is one global `EventSanitizer` bean. The opt-in
is binary โ sanitize, or pass through raw. If we later need per-listener
policy, that becomes a strategy choice inside the sanitizer rather than
another branch in `deliverEvent`.
##########
runtime/service/src/main/java/org/apache/polaris/service/events/EventSanitizer.java:
##########
@@ -0,0 +1,38 @@
+/*
+ * 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;
+
+/**
+ * Global, dispatcher-enforced sanitizer for events. Invoked exactly once at the choke point in
Review Comment:
Good catch โ fixed. Updated the Javadoc on `EventSanitizer` to read "Invoked
once per listener delivery from `PolarisEventListeners.deliverEvent`, before
each listener that has not opted in via `RawEventAccess`. Each invocation
produces a fresh sanitized copy; the original event is never mutated."
Also dropped the "future raw-event escape-hatch" wording from the second
paragraph since `RawEventAccess` now exists (see reply on
`PolarisEventListeners.java:84`).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Overhaul event persistence with universal routing and global secure-by-default sanitization [polaris]
Posted by "adutra (via GitHub)" <gi...@apache.org>.
adutra commented on PR #4225:
URL: https://github.com/apache/polaris/pull/4225#issuecomment-4662284712
> Is there any change that you would like me to do in this PR itself and not a follow-up? If not, can you approve this PR so that it can get merged and I will address your comments in a follow-up? Does that work?
Could you please create the GitHub issue for the `REALM_SCOPED` sentinel value, then tackle just this nit:
https://github.com/apache/polaris/pull/4225#discussion_r3379625418
After that I think we are good to merge.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Overhaul event persistence with universal routing and global secure-by-default sanitization [polaris]
Posted by "flyrain (via GitHub)" <gi...@apache.org>.
flyrain commented on PR #4225:
URL: https://github.com/apache/polaris/pull/4225#issuecomment-4663644234
Thanks @sririshindra for the PR. Thanks everyone for the review!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor persistence event listener to a universal registry pattern with strict allowlist serialization [polaris]
Posted by "sririshindra (via GitHub)" <gi...@apache.org>.
sririshindra commented on code in PR #4225:
URL: https://github.com/apache/polaris/pull/4225#discussion_r3231030623
##########
runtime/service/src/main/java/org/apache/polaris/service/events/listeners/DefaultPersistenceEventHandler.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * 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.listeners;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Objects;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.rest.requests.RenameTableRequest;
+import org.apache.polaris.core.admin.model.Catalog;
+import org.apache.polaris.core.admin.model.Principal;
+import org.apache.polaris.core.auth.PolarisPrincipal;
+import org.apache.polaris.core.entity.PolarisEvent.ResourceType;
+import org.apache.polaris.service.events.EventAttributeMap;
+import org.apache.polaris.service.events.EventAttributes;
+import org.apache.polaris.service.events.PolarisEvent;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+final class DefaultPersistenceEventHandler implements PersistenceEventHandler {
+
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(DefaultPersistenceEventHandler.class);
+ private static final String SERIALIZATION_ERROR_PAYLOAD = "{\"error\":\"Serialization failed\"}";
+
+ static final String ALL_CATALOGS_SCOPE = "_all_catalogs";
+ static final String ROOT_NAMESPACE_SCOPE = "_root_namespace";
+ static final String TABLE_SCOPE = "_table_scope";
+ static final String VIEW_SCOPE = "_view_scope";
+
+ private final PersistenceAttributeSecurityFilter securityFilter;
+ private final ObjectMapper objectMapper;
+
+ DefaultPersistenceEventHandler() {
+ this(new PersistenceAttributeSecurityFilter(), new ObjectMapper());
+ }
+
+ DefaultPersistenceEventHandler(
+ PersistenceAttributeSecurityFilter securityFilter, ObjectMapper objectMapper) {
+ this.securityFilter = Objects.requireNonNull(securityFilter, "securityFilter");
+ this.objectMapper = Objects.requireNonNull(objectMapper, "objectMapper");
+ }
+
+ @Override
+ public org.apache.polaris.core.entity.PolarisEvent toPersistenceEvent(PolarisEvent event) {
+ EventAttributeMap attributes = event.attributes();
+ String catalogName = resolveCatalogName(attributes);
+ ResourceType resourceType = resolveResourceType(event, attributes);
+
+ org.apache.polaris.core.entity.PolarisEvent persistenceEvent =
+ new org.apache.polaris.core.entity.PolarisEvent(
+ catalogName,
+ event.metadata().eventId().toString(),
+ event.metadata().requestId().orElse(null),
+ event.type().name(),
+ event.metadata().timestamp().toEpochMilli(),
+ event.metadata().user().map(PolarisPrincipal::getName).orElse(null),
+ resourceType,
+ resolveResourceIdentifier(resourceType, attributes, catalogName));
+
+ Map<String, String> additionalProperties =
+ new LinkedHashMap<>(securityFilter.filterAllowlistedAttributes(attributes));
+ additionalProperties.putAll(
+ securityFilter.filterAllowlistedTelemetryContext(event.metadata().openTelemetryContext()));
+ if (!additionalProperties.isEmpty()) {
+ persistenceEvent.setAdditionalProperties(
+ serializeAdditionalProperties(event, additionalProperties));
+ }
+
+ return persistenceEvent;
+ }
+
+ private String serializeAdditionalProperties(
+ PolarisEvent event, Map<String, String> additionalProperties) {
+ try {
+ return objectMapper.writeValueAsString(additionalProperties);
+ } catch (JsonProcessingException e) {
+ LOGGER.warn(
+ "Failed to serialize additional properties for eventType={} eventId={}; persisting fallback payload",
+ event.type(),
+ event.metadata().eventId());
+ LOGGER.debug(
+ "Serialization failure details for eventType={} eventId={}",
+ event.type(),
+ event.metadata().eventId(),
+ e);
+ return SERIALIZATION_ERROR_PAYLOAD;
+ }
+ }
+
+ private static String resolveCatalogName(EventAttributeMap attributes) {
+ return attributes
+ .get(EventAttributes.CATALOG_NAME)
+ .or(() -> attributes.get(EventAttributes.CATALOG).map(Catalog::getName))
+ .orElse(ALL_CATALOGS_SCOPE);
+ }
+
+ private static ResourceType resolveResourceType(
+ PolarisEvent event, EventAttributeMap attributes) {
+ return switch (event.type().category()) {
+ case CATALOG -> ResourceType.CATALOG;
+ case NAMESPACE -> ResourceType.NAMESPACE;
+ case TABLE -> ResourceType.TABLE;
+ case VIEW -> ResourceType.VIEW;
+ // Persistence resource type schema supports CATALOG/NAMESPACE/TABLE/VIEW only.
Review Comment:
All event types are now persisted unconditionally. ResourceType is a 4-value classification field (CATALOG/NAMESPACE/TABLE/VIEW) used for indexing โ not a filtering gate. Every event that reaches the listener is mapped to one of these types via PolarisEventType.category() and persisted. Event type filtering remains configurable at the listener registration level in PolarisEventListenerConfiguration for operators who want a subset.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org