(polaris) branch main updated: Give the in-memory event buffer its own lock (#4572) (#4577)

[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 d9646354e9 Give the in-memory event buffer its own lock (#4572) (#4577)
d9646354e9 is described below

commit d9646354e9023630c7a238687087be05b7a3e5f0
Author: Rahul Joshi <vi...@gmail.com>
AuthorDate: Thu Jun 11 04:33:14 2026 -0400

    Give the in-memory event buffer its own lock (#4572) (#4577)
    
    Replace the borrowed-monitor approach from #4498 with an EventProcessor
    that holds its own ReentrantLock and a completed flag, so onNext and
    onComplete no longer depend on UnicastProcessor's internal synchronized.
    processEvent retries on eviction; shutdown uses a read/write lock to
    drain in-flight events. Removes the completeSynchronized helper.
---
 .../inmemory/InMemoryBufferEventListener.java      | 135 +++++++++++++++------
 .../InMemoryBufferEventListenerBufferSizeTest.java |   6 +-
 .../InMemoryBufferEventListenerTestBase.java       |  10 +-
 3 files changed, 109 insertions(+), 42 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 8a6aee707a..c2144c3f37 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
@@ -32,6 +32,8 @@ import jakarta.inject.Inject;
 import java.time.Duration;
 import java.util.List;
 import java.util.Objects;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
 import org.apache.polaris.core.PolarisCallContext;
 import org.apache.polaris.core.context.RealmContext;
 import org.apache.polaris.core.entity.PolarisEvent;
@@ -51,58 +53,117 @@ public class InMemoryBufferEventListener extends PolarisPersistenceEventListener
   @Inject MetaStoreManagerFactory metaStoreManagerFactory;
   @Inject InMemoryBufferEventListenerConfiguration configuration;
 
+  /**
+   * Thrown by {@link EventProcessor#onNext} when the wrapped processor has already completed (for
+   * example, it was evicted between the cache lookup and the {@code onNext} call). {@link
+   * #processEvent} catches it and retries with a freshly loaded processor.
+   */
+  private static final class CompletedException extends IllegalStateException {}
+
+  /**
+   * Wraps a {@link UnicastProcessor} together with its own lock, so that mutual exclusion between
+   * {@code onNext} (from {@link #processEvent}) and {@code onComplete} (from eviction or shutdown)
+   * does not depend on smallrye-mutiny's internal {@code synchronized} on {@code
+   * UnicastProcessor.onNext}.
+   */
+  protected final class EventProcessor {
+
+    @VisibleForTesting final UnicastProcessor<PolarisEvent> processor;
+    private final ReentrantLock lock = new ReentrantLock();
+    private boolean completed = false; // guarded by lock
+
+    EventProcessor(String realmId) {
+      processor = UnicastProcessor.create();
+      processor
+          .emitOn(Infrastructure.getDefaultWorkerPool())
+          .group()
+          .intoLists()
+          .of(configuration.maxBufferSize(), configuration.bufferTime())
+          .subscribe()
+          .with(events -> flush(realmId, events), error -> onProcessorError(realmId, error));
+    }
+
+    void onNext(PolarisEvent event) {
+      lock.lock();
+      try {
+        if (completed) {
+          throw new CompletedException();
+        }
+        processor.onNext(event);
+      } finally {
+        lock.unlock();
+      }
+    }
+
+    void onComplete() {
+      lock.lock();
+      try {
+        if (!completed) {
+          completed = true;
+          processor.onComplete();
+        }
+      } finally {
+        lock.unlock();
+      }
+    }
+  }
+
   @VisibleForTesting
-  final LoadingCache<String, UnicastProcessor<PolarisEvent>> processors =
+  final LoadingCache<String, EventProcessor> processors =
       Caffeine.newBuilder()
           .expireAfterAccess(Duration.ofHours(1))
           .evictionListener(
-              (String realmId, UnicastProcessor<?> processor, RemovalCause cause) ->
-                  completeSynchronized(processor))
+              (String realmId, EventProcessor processor, RemovalCause cause) -> {
+                if (processor != null) {
+                  processor.onComplete();
+                }
+              })
           .build(this::createProcessor);
 
+  // Construct via a method (not EventProcessor::new) so that when the cache lives on a CDI client
+  // proxy, the call is delegated to the contextual bean instance, whose injected configuration is
+  // non-null. A direct inner-class instantiation would capture the proxy as the enclosing instance
+  // and read its uninjected (null) configuration.
+  protected EventProcessor createProcessor(String realmId) {
+    return new EventProcessor(realmId);
+  }
+
+  private final ReentrantReadWriteLock shutdownLock = new ReentrantReadWriteLock();
+  private boolean shutdown = false; // guarded by shutdownLock
+
   @Override
   protected void processEvent(String realmId, PolarisEvent event) {
-    var processor = Objects.requireNonNull(processors.get(realmId));
-    // UnicastProcessor.onNext() is internally synchronized (smallrye-mutiny
-    // UnicastProcessor declares onNext as `public synchronized void`), so concurrent
-    // processEvent() calls for the same realm serialize on the processor's intrinsic
-    // lock without an external guard.
-    processor.onNext(event);
+    shutdownLock.readLock().lock();
+    try {
+      if (shutdown) {
+        return;
+      }
+      while (true) {
+        var processor = Objects.requireNonNull(processors.get(realmId));
+        try {
+          processor.onNext(event);
+          return;
+        } catch (CompletedException ignored) {
+          // processor was evicted between the cache lookup and onNext; retry with a fresh one
+        }
+      }
+    } finally {
+      shutdownLock.readLock().unlock();
+    }
   }
 
   @PreDestroy
   public void shutdown() {
-    processors.asMap().values().forEach(InMemoryBufferEventListener::completeSynchronized);
-    processors.invalidateAll(); // doesn't call the eviction listener
-  }
-
-  /**
-   * Calls {@link UnicastProcessor#onComplete()} while holding the processor's intrinsic monitor.
-   *
-   * <p>smallrye-mutiny's {@code UnicastProcessor.onNext} is method-{@code synchronized} on the
-   * processor instance; {@code onComplete} is not. Acquiring the same intrinsic monitor here
-   * restores symmetric mutual exclusion between concurrent {@code onNext} (from {@code
-   * processEvent}) and {@code onComplete} (from eviction or shutdown). Wrapping the pattern as an
-   * invariant keeps the synchronization requirement structurally visible to future maintainers.
-   */
-  private static void completeSynchronized(UnicastProcessor<?> processor) {
-    synchronized (processor) {
-      processor.onComplete();
+    shutdownLock.writeLock().lock();
+    try {
+      shutdown = true;
+      processors.asMap().values().forEach(EventProcessor::onComplete);
+      processors.invalidateAll(); // doesn't call the eviction listener
+    } finally {
+      shutdownLock.writeLock().unlock();
     }
   }
 
-  protected UnicastProcessor<PolarisEvent> createProcessor(String realmId) {
-    UnicastProcessor<PolarisEvent> processor = UnicastProcessor.create();
-    processor
-        .emitOn(Infrastructure.getDefaultWorkerPool())
-        .group()
-        .intoLists()
-        .of(configuration.maxBufferSize(), configuration.bufferTime())
-        .subscribe()
-        .with(events -> flush(realmId, events), error -> onProcessorError(realmId, error));
-    return processor;
-  }
-
   @Retry(maxRetries = 5, delay = 1000, jitter = 100)
   @Fallback(fallbackMethod = "onFlushError")
   protected void flush(String realmId, List<PolarisEvent> events) {
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 7ee12ebc02..aa80ad138a 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
@@ -28,10 +28,8 @@ import com.google.common.collect.ImmutableMap;
 import io.quarkus.test.junit.QuarkusTest;
 import io.quarkus.test.junit.QuarkusTestProfile;
 import io.quarkus.test.junit.TestProfile;
-import io.smallrye.mutiny.operators.multi.processors.UnicastProcessor;
 import io.smallrye.mutiny.subscription.BackPressureFailure;
 import java.util.Map;
-import org.apache.polaris.core.entity.PolarisEvent;
 import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestInstance;
@@ -89,10 +87,10 @@ class InMemoryBufferEventListenerBufferSizeTest extends InMemoryBufferEventListe
   @Test
   void testProcessorFailureRecovery() {
     producer.processEvent("test1", event());
-    UnicastProcessor<PolarisEvent> test1 = producer.processors.get("test1");
+    var test1 = producer.processors.get("test1");
     assertThat(test1).isNotNull();
     // emulate backpressure error; will drop the event and invalidate the processor
-    test1.onError(new BackPressureFailure("error"));
+    test1.processor.onError(new BackPressureFailure("error"));
     // will create a new processor and recover
     sendAsync("test1", 10);
     assertRows("test1", 10);
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/events/listeners/inmemory/InMemoryBufferEventListenerTestBase.java b/runtime/service/src/test/java/org/apache/polaris/service/events/listeners/inmemory/InMemoryBufferEventListenerTestBase.java
index d86fb44e8a..a1d9248519 100644
--- a/runtime/service/src/test/java/org/apache/polaris/service/events/listeners/inmemory/InMemoryBufferEventListenerTestBase.java
+++ b/runtime/service/src/test/java/org/apache/polaris/service/events/listeners/inmemory/InMemoryBufferEventListenerTestBase.java
@@ -41,6 +41,7 @@ import javax.sql.DataSource;
 import org.apache.polaris.core.entity.PolarisEvent;
 import org.apache.polaris.core.persistence.MetaStoreManagerFactory;
 import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
 
 abstract class InMemoryBufferEventListenerTestBase {
 
@@ -64,6 +65,8 @@ abstract class InMemoryBufferEventListenerTestBase {
 
   @Inject
   @Identifier("persistence-in-memory-buffer")
+  Instance<InMemoryBufferEventListener> producerInstance;
+
   InMemoryBufferEventListener producer;
 
   @InjectSpy
@@ -78,10 +81,15 @@ abstract class InMemoryBufferEventListenerTestBase {
 
   @Inject Instance<DataSource> dataSource;
 
+  @BeforeEach
+  void resolveProducer() {
+    producer = producerInstance.get();
+  }
+
   @AfterEach
   void clearEvents() throws Exception {
     reset(metaStoreManagerFactory);
-    producer.shutdown();
+    producerInstance.destroy(producer);
     try (Connection connection = dataSource.get().getConnection();
         Statement statement = connection.createStatement()) {
       statement.execute("DELETE FROM polaris_schema.events");