Introducing changes to enable forwarding of Polaris events to AWS Cloudwatch. #2962
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
adutra
reviewed
Nov 3, 2025
| # polaris.event-listener.aws-cloudwatch.log-stream=polaris-cloudwatch-default-stream | ||
| # polaris.event-listener.aws-cloudwatch.region=us-east-1 | ||
| # polaris.event-listener.aws-cloudwatch.synchronous-mode=false | ||
| # polaris.event-listener.aws-cloudwatch.event-types= // the absence of this property would result if processing all Polaris event types. |
There was a problem hiding this comment.
Suggested change
| # polaris.event-listener.aws-cloudwatch.event-types= // the absence of this property would result if processing all Polaris event types. | |
| # polaris.event-listener.aws-cloudwatch.event-types= // the absence of this property would result in processing all Polaris event types. |
| executorService = Executors.newSingleThreadExecutor(); | ||
|
|
||
| // Build the test mapper and apply the same customizations Quarkus would | ||
| objectMapper = new ObjectMapper(); |
There was a problem hiding this comment.
nit: could be a final field.
| AwsCloudWatchConfiguration config, | ||
| Clock clock, | ||
| PolarisIcebergObjectMapperCustomizer customizer) { | ||
| PolarisIcebergObjectMapperCustomizer customizer, |
There was a problem hiding this comment.
You don't need to pass the customizer, the customizer will be invoked for you by Quarkus. Just inject the object mapper.
| protected abstract void handle(PolarisEvent event); | ||
|
|
||
| /** Optional filter (config-based). Default: handle all. */ | ||
| protected boolean shouldHandle(Object event) { |
There was a problem hiding this comment.
Suggested change
| protected boolean shouldHandle(Object event) { | |
| protected boolean shouldHandle(PolarisEvent event) { |
Comment on lines
+90
to
+94
| this.listenToAllEvents = | ||
| config.eventTypes().isEmpty() | ||
| || config.eventTypes().map(Set::isEmpty).orElse(true) | ||
| || config.eventTypes().get().stream().anyMatch(e -> e == PolarisEvent.class); | ||
| this.allowedEventTypes = listenToAllEvents ? Set.of() : Set.copyOf(config.eventTypes().get()); |
There was a problem hiding this comment.
nit: I think we can simplify a bit:
this.allowedEventTypes = config.eventTypes().orElse(Set.of());
this.listenToAllEvents =
allowedEventTypes.isEmpty()
|| allowedEventTypes.stream().anyMatch(c -> c == PolarisEvent.class);| if (this.listenToAllEvents) { | ||
| return true; | ||
| } | ||
| Class<? extends PolarisEvent> actualType = polarisEvent.getClass(); |
There was a problem hiding this comment.
Note: this is probably a bit inefficient under a high load. Switching to event type IDs should help because the lookup would be faster when using an EnumSet.
| } | ||
|
|
||
| @PostConstruct | ||
| void verifyMapper() { |
| import org.apache.polaris.service.events.PrincipalsServiceEvents; | ||
|
|
||
| /** | ||
| * Base class for event listeners that with to generically forward all {@link PolarisEvent |
There was a problem hiding this comment.
Suggested change
| * Base class for event listeners that with to generically forward all {@link PolarisEvent | |
| * Base class for event listeners that wish to generically forward all {@link PolarisEvent |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
These changes were created to address concerns raised in Issues #2630
What changes were proposed in this pull request?
Streamlining Event-to-JSON Serialization:
The previous architecture involved a redundant, two-step conversion process: events were transformed into intermediate Maps by an abstract class, and only then was the resulting map serialized into JSON (as seen in AwsCloudWatchEventListener).
Since Jackson JSON is our chosen serialization format, we've removed this unnecessary intermediate mapping step. We introduced Jackson Mixins to provide native JSON serialization support directly on the event objects. This refactoring simplifies the serialization pipeline from:
Event -> Map -> JSON
to
Event -> JSON
Introduces
AllEventsForwardingListener, a generic event listener designed to automatically receive and forward all Polaris events without requiring explicit event-type configuration.Introduces
polaris.event-listener.aws-cloudwatch.event-typesproperty: This property introduces a configurable event filter for the AWS CloudWatch event listener.It defines which Polaris event types should be serialized and forwarded to CloudWatch Logs.
Why are the changes needed?
Does this PR introduce any user-facing change?
How was this patch tested?
CHANGELOG.md