(polaris) branch main updated: InMemoryBufferEventListener: fix cache removal logic and flaky test (#4677)

[ad...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

adutra pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/polaris.git


The following commit(s) were added to refs/heads/main by this push:
     new 5f219901b3 InMemoryBufferEventListener: fix cache removal logic and flaky test (#4677)
5f219901b3 is described below

commit 5f219901b3838723bcde5540dd624cb9558c35eb
Author: Alexandre Dutra <ad...@apache.org>
AuthorDate: Thu Jun 11 17:18:38 2026 +0200

    InMemoryBufferEventListener: fix cache removal logic and flaky test (#4677)
    
    This PR fixes one bug and one flaky test:
    
    - `InMemoryBufferEventListener`'s internal cache was using an eviction
    listener to complete the failed processor. The eviction listener is not
    the right one: it doesn't trigger when the entry is evicted explicitly.
    The right listener to use is a removal listener. It's not a big deal
    because explicit removals generally mean that the processor errored out,
    or that the application is shutting down; but it's better to go through
    the listener in all cases.
    
    - The test `testProcessorFailureRecovery()` was flaky for two reasons:
    it was accessing a field on a CDI bean, which is not correct; and it
    wasn't waiting for the processor to be evicted before sending more
    events.
---
 .../inmemory/InMemoryBufferEventListener.java          | 18 +++++++++++++-----
 .../InMemoryBufferEventListenerBufferSizeTest.java     |  8 +++++++-
 2 files changed, 20 insertions(+), 6 deletions(-)

diff --git a/runtime/service/src/main/java/org/apache/polaris/service/events/listeners/inmemory/InMemoryBufferEventListener.java b/runtime/service/src/main/java/org/apache/polaris/service/events/listeners/inmemory/InMemoryBufferEventListener.java
index c2144c3f37..8c272217d9 100644
--- a/runtime/service/src/main/java/org/apache/polaris/service/events/listeners/inmemory/InMemoryBufferEventListener.java
+++ b/runtime/service/src/main/java/org/apache/polaris/service/events/listeners/inmemory/InMemoryBufferEventListener.java
@@ -108,11 +108,10 @@ public class InMemoryBufferEventListener extends PolarisPersistenceEventListener
     }
   }
 
-  @VisibleForTesting
-  final LoadingCache<String, EventProcessor> processors =
+  private final LoadingCache<String, EventProcessor> processors =
       Caffeine.newBuilder()
           .expireAfterAccess(Duration.ofHours(1))
-          .evictionListener(
+          .removalListener(
               (String realmId, EventProcessor processor, RemovalCause cause) -> {
                 if (processor != null) {
                   processor.onComplete();
@@ -157,8 +156,7 @@ public class InMemoryBufferEventListener extends PolarisPersistenceEventListener
     shutdownLock.writeLock().lock();
     try {
       shutdown = true;
-      processors.asMap().values().forEach(EventProcessor::onComplete);
-      processors.invalidateAll(); // doesn't call the eviction listener
+      processors.invalidateAll();
     } finally {
       shutdownLock.writeLock().unlock();
     }
@@ -187,4 +185,14 @@ public class InMemoryBufferEventListener extends PolarisPersistenceEventListener
         error);
     processors.invalidate(realmId);
   }
+
+  @VisibleForTesting
+  EventProcessor processor(String realmId) {
+    return processors.get(realmId);
+  }
+
+  @VisibleForTesting
+  boolean hasProcessor(String realmId) {
+    return processors.getIfPresent(realmId) != null;
+  }
 }
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/events/listeners/inmemory/InMemoryBufferEventListenerBufferSizeTest.java b/runtime/service/src/test/java/org/apache/polaris/service/events/listeners/inmemory/InMemoryBufferEventListenerBufferSizeTest.java
index aa80ad138a..ca6bcbdd84 100644
--- a/runtime/service/src/test/java/org/apache/polaris/service/events/listeners/inmemory/InMemoryBufferEventListenerBufferSizeTest.java
+++ b/runtime/service/src/test/java/org/apache/polaris/service/events/listeners/inmemory/InMemoryBufferEventListenerBufferSizeTest.java
@@ -20,6 +20,7 @@
 package org.apache.polaris.service.events.listeners.inmemory;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.doThrow;
@@ -29,6 +30,7 @@ import io.quarkus.test.junit.QuarkusTest;
 import io.quarkus.test.junit.QuarkusTestProfile;
 import io.quarkus.test.junit.TestProfile;
 import io.smallrye.mutiny.subscription.BackPressureFailure;
+import java.time.Duration;
 import java.util.Map;
 import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
 import org.junit.jupiter.api.Test;
@@ -87,10 +89,14 @@ class InMemoryBufferEventListenerBufferSizeTest extends InMemoryBufferEventListe
   @Test
   void testProcessorFailureRecovery() {
     producer.processEvent("test1", event());
-    var test1 = producer.processors.get("test1");
+    var test1 = producer.processor("test1");
     assertThat(test1).isNotNull();
     // emulate backpressure error; will drop the event and invalidate the processor
     test1.processor.onError(new BackPressureFailure("error"));
+    // wait for the processor to be evicted from the cache
+    await()
+        .atMost(Duration.ofSeconds(10))
+        .untilAsserted(() -> assertThat(producer.hasProcessor("test1")).isFalse());
     // will create a new processor and recover
     sendAsync("test1", 10);
     assertRows("test1", 10);