Revert "Includes the following changes: 1. Fixing a few types 2. removing the redundant usage of object mapper customizer in AwsCloudWatchEventListener and related changes."

This reverts commit 99d8a5d.

Author: vchag

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

@@ -145,7 +145,7 @@ polaris.event-listener.type=no-op
 # 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 in processing all Polaris event types.
+# polaris.event-listener.aws-cloudwatch.event-types= // the absence of this property would result if processing all Polaris event types.
 
 polaris.log.request-id-header-name=Polaris-Request-Id
 # polaris.log.mdc.aid=polaris

runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventType.java

@@ -31,7 +31,7 @@
 import org.apache.polaris.service.events.PrincipalsServiceEvents;
 
 /**
- * Base class for event listeners that wish to generically forward all {@link PolarisEvent
+ * Base class for event listeners that with to generically forward all {@link PolarisEvent
  * PolarisEvents} to an external sink.
  *
  * <p>This design follows the Template Method pattern, centralizing shared control flow in the base

runtime/service/src/main/java/org/apache/polaris/service/events/listeners/AllEventsForwardingListener.java

@@ -39,6 +39,7 @@
 import java.util.function.Supplier;
 import org.apache.polaris.core.auth.PolarisPrincipal;
 import org.apache.polaris.core.context.CallContext;
+import org.apache.polaris.service.config.PolarisIcebergObjectMapperCustomizer;
 import org.apache.polaris.service.events.PolarisEvent;
 import org.apache.polaris.service.events.listeners.AllEventsForwardingListener;
 import org.slf4j.Logger;
@@ -75,17 +76,22 @@ public class AwsCloudWatchEventListener extends AllEventsForwardingListener {
 
   @Inject
   public AwsCloudWatchEventListener(
-      AwsCloudWatchConfiguration config, Clock clock, ObjectMapper mapper) {
+      AwsCloudWatchConfiguration config,
+      Clock clock,
+      PolarisIcebergObjectMapperCustomizer customizer,
+      ObjectMapper mapper) {
     this.logStream = config.awsCloudWatchLogStream();
     this.logGroup = config.awsCloudWatchLogGroup();
     this.region = Region.of(config.awsCloudWatchRegion());
     this.synchronousMode = config.synchronousMode();
     this.clock = clock;
     this.objectMapper = mapper;
-    this.allowedEventTypes = config.eventTypes().orElse(Set.of());
+    customizer.customize(this.objectMapper);
     this.listenToAllEvents =
-        allowedEventTypes.isEmpty()
-            || allowedEventTypes.stream().anyMatch(c -> c == PolarisEvent.class);
+        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());
   }
 
   @Override
@@ -112,6 +118,14 @@ void start() {
     ensureLogGroupAndStream();
   }
 
+  @PostConstruct
+  void verifyMapper() {
+    LOGGER.info(
+        "ObjectMapper hash={}, mixins={}",
+        System.identityHashCode(objectMapper),
+        objectMapper.mixInCount());
+  }
+
   protected CloudWatchLogsAsyncClient createCloudWatchAsyncClient() {
     return CloudWatchLogsAsyncClient.builder().region(region).build();
   }

runtime/service/src/main/java/org/apache/polaris/service/events/listeners/aws/cloudwatch/AwsCloudWatchEventListener.java

@@ -31,7 +31,6 @@
 import java.math.BigInteger;
 import java.time.Clock;
 import java.time.Duration;
-import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -93,20 +92,22 @@ class AwsCloudWatchEventListenerTest {
 
   private ExecutorService executorService;
   private AutoCloseable mockitoContext;
-  private static final ObjectMapper objectMapper = new ObjectMapper();
+  private ObjectMapper objectMapper;
 
   @BeforeEach
   void setUp() {
     mockitoContext = MockitoAnnotations.openMocks(this);
     executorService = Executors.newSingleThreadExecutor();
-    customizer.customize(objectMapper);
+
+    // Build the test mapper and apply the same customizations Quarkus would
+    objectMapper = new ObjectMapper();
 
     // Configure the mocks
     when(config.awsCloudWatchLogGroup()).thenReturn(LOG_GROUP);
     when(config.awsCloudWatchLogStream()).thenReturn(LOG_STREAM);
     when(config.awsCloudWatchRegion()).thenReturn("us-east-1");
     when(config.synchronousMode()).thenReturn(false); // default async
-    when(config.eventTypes()).thenReturn(Optional.empty()); // handle all events
+    when(config.eventTypes()).thenReturn(java.util.Optional.empty()); // handle all events
   }
 
   @AfterEach
@@ -140,7 +141,7 @@ private CloudWatchLogsAsyncClient createCloudWatchAsyncClient() {
 
   private AwsCloudWatchEventListener createListener(CloudWatchLogsAsyncClient client) {
     AwsCloudWatchEventListener listener =
-        new AwsCloudWatchEventListener(config, clock, objectMapper) {
+        new AwsCloudWatchEventListener(config, clock, customizer, objectMapper) {
           @Override
           protected CloudWatchLogsAsyncClient createCloudWatchAsyncClient() {
             return client;
@@ -328,9 +329,7 @@ void shouldSendEventInSynchronousMode() {
   void ensureObjectMapperCustomizerIsApplied() {
 
     AwsCloudWatchEventListener listener =
-        new AwsCloudWatchEventListener(config, clock, objectMapper);
-
-    assertThat(listener.objectMapper).isSameAs(objectMapper);
+        new AwsCloudWatchEventListener(config, clock, customizer, objectMapper);
 
     assertThat(listener.objectMapper.getPropertyNamingStrategy())
         .isInstanceOf(PropertyNamingStrategies.KebabCaseStrategy.class);
@@ -353,10 +352,10 @@ void ensureObjectMapperCustomizerIsApplied() {
   @Test
   void shouldListenToAllEventTypesWhenConfigNotProvided() {
     // given: config.eventTypes() is empty → listen to all events
-    when(config.eventTypes()).thenReturn(Optional.empty());
+    when(config.eventTypes()).thenReturn(java.util.Optional.empty());
 
     AwsCloudWatchEventListener listener =
-        new AwsCloudWatchEventListener(config, clock, objectMapper);
+        new AwsCloudWatchEventListener(config, clock, customizer, objectMapper);
 
     // This is any random PolarisEvent — if the listener listens to all types,
     // shouldHandle(event) should return true