Move event emitting off the main thread to avoid deadlocks

When stacking event emitting inside an EventProvider, when using sychronization
the EventProvider can deadlock, to avoid this move the event emitting of the
main thread.

Signed-off-by: Philipp Fehre <philipp@fehre.co.uk>

Author: Philipp Fehre

src/main/java/dev/openfeature/sdk/EventProvider.java

@@ -1,6 +1,10 @@
 package dev.openfeature.sdk;
 
 import dev.openfeature.sdk.internal.TriConsumer;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
 
 /**
  * Abstract EventProvider. Providers must extend this class to support events.
@@ -14,8 +18,10 @@
  *
  * @see FeatureProvider
  */
+@Slf4j
 public abstract class EventProvider implements FeatureProvider {
     private EventProviderListener eventProviderListener;
+    private final ExecutorService emitterExecutor = Executors.newCachedThreadPool();
 
     void setEventProviderListener(EventProviderListener eventProviderListener) {
         this.eventProviderListener = eventProviderListener;
@@ -46,6 +52,24 @@ void detach() {
         this.onEmit = null;
     }
 
+    /**
+     * Stop the event emitter executor and block until either termination has completed
+     * or timeout period has elapsed.
+     */
+    @Override
+    public void shutdown() {
+        emitterExecutor.shutdown();
+        try {
+            if (!emitterExecutor.awaitTermination(EventSupport.SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
+                log.warn("Emitter executor did not terminate before the timeout period had elapsed");
+                emitterExecutor.shutdownNow();
+            }
+        } catch (InterruptedException e) {
+            emitterExecutor.shutdownNow();
+            Thread.currentThread().interrupt();
+        }
+    }
+
     /**
      * Emit the specified {@link ProviderEvent}.
      *
@@ -56,8 +80,10 @@ public void emit(ProviderEvent event, ProviderEventDetails details) {
         if (eventProviderListener != null) {
             eventProviderListener.onEmit(event, details);
         }
-        if (this.onEmit != null) {
-            this.onEmit.accept(this, event, details);
+
+        final TriConsumer<EventProvider, ProviderEvent, ProviderEventDetails> localOnEmit = this.onEmit;
+        if (localOnEmit != null) {
+            emitterExecutor.submit(() -> localOnEmit.accept(this, event, details));
         }
     }
 

src/main/java/dev/openfeature/sdk/EventSupport.java

@@ -19,15 +19,15 @@
 @Slf4j
 class EventSupport {
 
+    public static final int SHUTDOWN_TIMEOUT_SECONDS = 3;
+
     // we use a v4 uuid as a "placeholder" for anonymous clients, since
     // ConcurrentHashMap doesn't support nulls
     private static final String defaultClientUuid = UUID.randomUUID().toString();
-    private static final int SHUTDOWN_TIMEOUT_SECONDS = 3;
     private final Map<String, HandlerStore> handlerStores = new ConcurrentHashMap<>();
     private final HandlerStore globalHandlerStore = new HandlerStore();
     private final ExecutorService taskExecutor = Executors.newCachedThreadPool(runnable -> {
         final Thread thread = new Thread(runnable);
-        thread.setDaemon(true);
         return thread;
     });
 

src/test/java/dev/openfeature/sdk/EventProviderTest.java

@@ -2,13 +2,19 @@
 
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.*;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
 
 import dev.openfeature.sdk.internal.TriConsumer;
+import dev.openfeature.sdk.testutils.TestStackedEmitCallsProvider;
+import io.cucumber.java.AfterAll;
 import lombok.SneakyThrows;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
 
 class EventProviderTest {
 
@@ -21,6 +27,11 @@ void setup() {
         eventProvider.initialize(null);
     }
 
+    @AfterAll
+    public static void resetDefaultProvider() {
+        OpenFeatureAPI.getInstance().setProviderAndWait(new NoOpProvider());
+    }
+
     @Test
     @DisplayName("should run attached onEmit with emitters")
     void emitsEventsWhenAttached() {
@@ -75,6 +86,15 @@ void doesNotThrowWhenOnEmitSame() {
         eventProvider.attach(onEmit2); // should not throw, same instance. noop
     }
 
+    @Test
+    @SneakyThrows
+    @Timeout(value = 2, threadMode = Timeout.ThreadMode.SEPARATE_THREAD)
+    @DisplayName("should not deadlock on emit called during emit")
+    void doesNotDeadlockOnEmitStackedCalls() {
+        TestStackedEmitCallsProvider provider = new TestStackedEmitCallsProvider();
+        OpenFeatureAPI.getInstance().setProviderAndWait(provider);
+    }
+
     static class TestEventProvider extends EventProvider {
 
         private static final String NAME = "TestEventProvider";

src/test/java/dev/openfeature/sdk/EventsTest.java

@@ -19,7 +19,7 @@
 
 class EventsTest {
 
-    private static final int TIMEOUT = 300;
+    private static final int TIMEOUT = 500;
     private static final int INIT_DELAY = TIMEOUT / 2;
 
     @AfterAll

src/test/java/dev/openfeature/sdk/testutils/TestStackedEmitCallsProvider.java

@@ -0,0 +1,103 @@
+package dev.openfeature.sdk.testutils;
+
+import dev.openfeature.sdk.EvaluationContext;
+import dev.openfeature.sdk.EventProvider;
+import dev.openfeature.sdk.Metadata;
+import dev.openfeature.sdk.ProviderEvaluation;
+import dev.openfeature.sdk.ProviderEvent;
+import dev.openfeature.sdk.ProviderEventDetails;
+import dev.openfeature.sdk.Value;
+import java.util.function.Consumer;
+
+public class TestStackedEmitCallsProvider extends EventProvider {
+    private final NestedBlockingEmitter nestedBlockingEmitter = new NestedBlockingEmitter(this::onProviderEvent);
+
+    @Override
+    public Metadata getMetadata() {
+        return () -> getClass().getSimpleName();
+    }
+
+    @Override
+    public void initialize(EvaluationContext evaluationContext) throws Exception {
+        synchronized (nestedBlockingEmitter) {
+            nestedBlockingEmitter.init();
+            while (!nestedBlockingEmitter.isReady()) {
+                try {
+                    nestedBlockingEmitter.wait();
+                } catch (InterruptedException e) {
+                }
+            }
+        }
+    }
+
+    private void onProviderEvent(ProviderEvent providerEvent) {
+        synchronized (nestedBlockingEmitter) {
+            if (providerEvent == ProviderEvent.PROVIDER_READY) {
+                nestedBlockingEmitter.setReady();
+                /*
+                 * This line deadlocked in the original implementation without the emitterExecutor see
+                 * https://github.com/open-feature/java-sdk/issues/1299
+                 */
+                emitProviderReady(ProviderEventDetails.builder().build());
+            }
+        }
+    }
+
+    @Override
+    public ProviderEvaluation<Boolean> getBooleanEvaluation(String key, Boolean defaultValue, EvaluationContext ctx) {
+        throw new UnsupportedOperationException("Unimplemented method 'getBooleanEvaluation'");
+    }
+
+    @Override
+    public ProviderEvaluation<String> getStringEvaluation(String key, String defaultValue, EvaluationContext ctx) {
+        throw new UnsupportedOperationException("Unimplemented method 'getStringEvaluation'");
+    }
+
+    @Override
+    public ProviderEvaluation<Integer> getIntegerEvaluation(String key, Integer defaultValue, EvaluationContext ctx) {
+        throw new UnsupportedOperationException("Unimplemented method 'getIntegerEvaluation'");
+    }
+
+    @Override
+    public ProviderEvaluation<Double> getDoubleEvaluation(String key, Double defaultValue, EvaluationContext ctx) {
+        throw new UnsupportedOperationException("Unimplemented method 'getDoubleEvaluation'");
+    }
+
+    @Override
+    public ProviderEvaluation<Value> getObjectEvaluation(String key, Value defaultValue, EvaluationContext ctx) {
+        throw new UnsupportedOperationException("Unimplemented method 'getObjectEvaluation'");
+    }
+
+    static class NestedBlockingEmitter {
+
+        private final Consumer<ProviderEvent> emitProviderEvent;
+        private volatile boolean isReady;
+
+        public NestedBlockingEmitter(Consumer<ProviderEvent> emitProviderEvent) {
+            this.emitProviderEvent = emitProviderEvent;
+        }
+
+        public void init() {
+            // run init outside monitored thread
+            new Thread(() -> {
+                        try {
+                            Thread.sleep(500);
+                        } catch (InterruptedException e) {
+                            throw new RuntimeException(e);
+                        }
+
+                        emitProviderEvent.accept(ProviderEvent.PROVIDER_READY);
+                    })
+                    .start();
+        }
+
+        public boolean isReady() {
+            return isReady;
+        }
+
+        public synchronized void setReady() {
+            isReady = true;
+            this.notifyAll();
+        }
+    }
+}