fix: await shutdown in in-process mode

providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/grpc/GrpcConnector.java

@@ -89,14 +89,16 @@ public void shutdown() throws Exception {
         }
 
         try {
-            if (this.channel != null) {
+            if (this.channel != null && !this.channel.isShutdown()) {
                 this.channel.shutdown();
-                this.channel.awaitTermination(5, TimeUnit.SECONDS);
+                this.channel.awaitTermination(this.deadline, TimeUnit.MILLISECONDS);
             }
         } finally {
             this.cache.clear();
-            if (this.channel != null) {
+            if (this.channel != null && !this.channel.isShutdown()) {
                 this.channel.shutdownNow();
+                this.channel.awaitTermination(5000, TimeUnit.MILLISECONDS);
+                log.warn(String.format("Unable to shut down channel by %d deadline", this.deadline));
             }
         }
     }

providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/process/InProcessResolver.java

@@ -86,8 +86,9 @@ public void init() throws Exception {
 
     /**
      * Shutdown in-process resolver.
-     */
-    public void shutdown() {
+     * @throws InterruptedException if stream can't be closed within deadline.
+     */ 
+    public void shutdown() throws InterruptedException {
         flagStore.shutdown();
         this.connected.set(false);
     }

providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/FlagStore.java

@@ -56,8 +56,9 @@ public void init() {
 
     /**
      * Shutdown storage layer.
+     * @throws InterruptedException if stream can't be closed within deadline.
      */
-    public void shutdown() {
+    public void shutdown() throws InterruptedException {
         if (shutdown.getAndSet(true)) {
             return;
         }

providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/Storage.java

@@ -10,7 +10,7 @@
 public interface Storage {
     void init();
 
-    void shutdown();
+    void shutdown() throws InterruptedException;
 
     FeatureFlag getFlag(final String key);
 

providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/connector/Connector.java

@@ -11,5 +11,5 @@ public interface Connector {
 
     BlockingQueue<StreamPayload> getStream();
 
-    void shutdown();
+    void shutdown() throws InterruptedException;
 }

providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/connector/grpc/GrpcStreamConnector.java

@@ -1,5 +1,12 @@
 package dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.grpc;
 
+import java.util.Random;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.logging.Level;
+
 import dev.openfeature.contrib.providers.flagd.FlagdOptions;
 import dev.openfeature.contrib.providers.flagd.resolver.common.ChannelBuilder;
 import dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.Connector;
@@ -11,12 +18,6 @@
 import io.grpc.ManagedChannel;
 import lombok.extern.java.Log;
 
-import java.util.Random;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.logging.Level;
-
 /**
  * Implements the {@link Connector} contract and emit flags obtained from flagd sync gRPC contract.
  */
@@ -36,10 +37,17 @@ public class GrpcStreamConnector implements Connector {
 
     private final ManagedChannel channel;
     private final FlagSyncServiceGrpc.FlagSyncServiceStub serviceStub;
+    private final int deadline;
 
+    /**
+     * Construct a new GrpcStreamConnector.
+     * 
+     * @param options flagd options
+     */
     public GrpcStreamConnector(final FlagdOptions options) {
         channel = ChannelBuilder.nettyChannel(options);
         serviceStub = FlagSyncServiceGrpc.newStub(channel);
+        this.deadline = options.getDeadline();
     }
 
     /**
@@ -67,13 +75,25 @@ public BlockingQueue<StreamPayload> getStream() {
 
     /**
      * Shutdown gRPC stream connector.
+     * @throws InterruptedException if stream can't be closed within deadline.
      */
-    public void shutdown() {
+    public void shutdown() throws InterruptedException {
         if (shutdown.getAndSet(true)) {
             return;
         }
 
-        channel.shutdown();
+        try {
+            if (this.channel != null && !this.channel.isShutdown()) {
+                this.channel.shutdown();
+                this.channel.awaitTermination(this.deadline, TimeUnit.MILLISECONDS);
+            }
+        } finally {
+            if (this.channel != null && !this.channel.isShutdown()) {
+                this.channel.shutdownNow();
+                this.channel.awaitTermination(this.deadline, TimeUnit.MILLISECONDS);
+                log.warning(String.format("Unable to shut down channel by %d deadline", this.deadline));
+            }
+        }
     }
 
     /**

providers/flagd/src/test/java/dev/openfeature/contrib/providers/flagd/e2e/RunFlagdInProcessCucumberTest.java

@@ -1,5 +1,6 @@
 package dev.openfeature.contrib.providers.flagd.e2e;
 
+import org.junit.jupiter.api.Order;
 import org.junit.platform.suite.api.ConfigurationParameter;
 import org.junit.platform.suite.api.IncludeEngines;
 import org.junit.platform.suite.api.SelectClasspathResource;
@@ -8,6 +9,7 @@
 import static io.cucumber.junit.platform.engine.Constants.PLUGIN_PROPERTY_NAME;
 import static io.cucumber.junit.platform.engine.Constants.GLUE_PROPERTY_NAME;
 
+@Order(value = Integer.MAX_VALUE)
 @Suite
 @IncludeEngines("cucumber")
 @SelectClasspathResource("features/evaluation.feature")

providers/flagd/src/test/java/dev/openfeature/contrib/providers/flagd/e2e/RunFlagdRpcCucumberTest.java

@@ -1,5 +1,6 @@
 package dev.openfeature.contrib.providers.flagd.e2e;
 
+import org.apache.logging.log4j.core.config.Order;
 import org.junit.platform.suite.api.ConfigurationParameter;
 import org.junit.platform.suite.api.IncludeEngines;
 import org.junit.platform.suite.api.SelectClasspathResource;
@@ -8,6 +9,7 @@
 import static io.cucumber.junit.platform.engine.Constants.PLUGIN_PROPERTY_NAME;
 import static io.cucumber.junit.platform.engine.Constants.GLUE_PROPERTY_NAME;
 
+@Order(value = Integer.MAX_VALUE)
 @Suite
 @IncludeEngines("cucumber")
 @SelectClasspathResource("features/evaluation.feature")

providers/flagd/src/test/java/dev/openfeature/contrib/providers/flagd/e2e/process/FlagdInProcessSetup.java

@@ -1,24 +1,29 @@
 package dev.openfeature.contrib.providers.flagd.e2e.process;
 
+import org.junit.jupiter.api.Order;
+import org.junit.jupiter.api.parallel.Isolated;
+
 import dev.openfeature.contrib.providers.flagd.Config;
 import dev.openfeature.contrib.providers.flagd.FlagdOptions;
 import dev.openfeature.contrib.providers.flagd.FlagdProvider;
 import dev.openfeature.contrib.providers.flagd.e2e.steps.StepDefinitions;
-import dev.openfeature.sdk.Client;
-import dev.openfeature.sdk.OpenFeatureAPI;
+import dev.openfeature.sdk.FeatureProvider;
 import io.cucumber.java.BeforeAll;
 
+@Isolated()
+@Order(value = Integer.MAX_VALUE)
 public class FlagdInProcessSetup {
+
+    private static FeatureProvider provider;
 
     @BeforeAll()
     public static void setup() throws InterruptedException {
-        FlagdProvider provider = new FlagdProvider(FlagdOptions.builder()
+        FlagdInProcessSetup.provider = new FlagdProvider(FlagdOptions.builder()
         .resolverType(Config.Evaluator.IN_PROCESS)
+        .deadline(3000)
         .host("localhost")
         .port(9090)
         .build());
-        OpenFeatureAPI.getInstance().setProviderAndWait("process", provider);
-        Client client = OpenFeatureAPI.getInstance().getClient("process");
-        StepDefinitions.setClient(client);
+        StepDefinitions.setProvider(provider);
     }
 }

providers/flagd/src/test/java/dev/openfeature/contrib/providers/flagd/e2e/rpc/FlagdRpcSetup.java

@@ -1,25 +1,28 @@
 package dev.openfeature.contrib.providers.flagd.e2e.rpc;
 
+import org.junit.jupiter.api.Order;
+import org.junit.jupiter.api.parallel.Isolated;
+
 import dev.openfeature.contrib.providers.flagd.Config;
 import dev.openfeature.contrib.providers.flagd.FlagdOptions;
 import dev.openfeature.contrib.providers.flagd.FlagdProvider;
 import dev.openfeature.contrib.providers.flagd.e2e.steps.StepDefinitions;
-import dev.openfeature.sdk.Client;
-import dev.openfeature.sdk.OpenFeatureAPI;
+import dev.openfeature.sdk.FeatureProvider;
 import io.cucumber.java.BeforeAll;
 
-
+@Isolated()
+@Order(value = Integer.MAX_VALUE)
 public class FlagdRpcSetup {
 
+    private static FeatureProvider provider;
+
     @BeforeAll()
     public static void setup() {
-        FlagdProvider provider = new FlagdProvider(FlagdOptions.builder()
+        FlagdRpcSetup.provider = new FlagdProvider(FlagdOptions.builder()
         .resolverType(Config.Evaluator.RPC)
         // set a generous deadline, to prevent timeouts in actions
         .deadline(3000)
         .build());
-        OpenFeatureAPI.getInstance().setProvider("rpc", provider);
-        Client client = OpenFeatureAPI.getInstance().getClient("rpc");
-        StepDefinitions.setClient(client);
+        StepDefinitions.setProvider(provider);
     }
 }

providers/flagd/src/test/java/dev/openfeature/contrib/providers/flagd/e2e/steps/StepDefinitions.java

@@ -4,17 +4,22 @@
 
 import java.util.HashMap;
 import java.util.Map;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Order;
+import org.junit.jupiter.api.parallel.Isolated;
 
 import dev.openfeature.sdk.Client;
 import dev.openfeature.sdk.EvaluationContext;
+import dev.openfeature.sdk.FeatureProvider;
 import dev.openfeature.sdk.FlagEvaluationDetails;
 import dev.openfeature.sdk.ImmutableContext;
 import dev.openfeature.sdk.ImmutableStructure;
+import dev.openfeature.sdk.OpenFeatureAPI;
 import dev.openfeature.sdk.Reason;
 import dev.openfeature.sdk.Structure;
 import dev.openfeature.sdk.Value;
-import io.cucumber.java.BeforeAll;
+import io.cucumber.java.AfterAll;
 import io.cucumber.java.en.And;
 import io.cucumber.java.en.Given;
 import io.cucumber.java.en.Then;
@@ -23,10 +28,12 @@
 /**
  * Common test suite used by both RPC and in-process flagd providers.
  */
+@Isolated()
+@Order(value = Integer.MAX_VALUE)
 public class StepDefinitions {
 
-    private static final ReentrantReadWriteLock sync = new ReentrantReadWriteLock();
     private static Client client;
+    private static FeatureProvider provider;
 
     private String booleanFlagKey;
     private String stringFlagKey;
@@ -63,15 +70,25 @@ public class StepDefinitions {
      * 
      * @param client client to inject into test.
      */
-    public static void setClient(Client client) {
-        StepDefinitions.client = client;
+    public static void setProvider(FeatureProvider provider) {
+        StepDefinitions.provider = provider;
     }
 
-    @BeforeAll()
+    @BeforeEach()
     @Given("a provider is registered")
     @Given("a flagd provider is set")
     public static void setup() {
-        // this is handled by the "Setup" files
+        if (StepDefinitions.client == null) {
+            OpenFeatureAPI.getInstance().setProviderAndWait("e2e", provider);
+            StepDefinitions.client = OpenFeatureAPI.getInstance().getClient("e2e");
+        }
+    }
+
+    @AfterAll()
+    public static void cleanUp() throws InterruptedException {
+        StepDefinitions.provider.shutdown();
+        StepDefinitions.provider = null;
+        StepDefinitions.client = null;
     }
 
     /*

providers/flagd/src/test/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/FlagStoreTest.java

@@ -18,7 +18,7 @@
 class FlagStoreTest {
 
     @Test
-    public void connectorHandling() {
+    public void connectorHandling() throws InterruptedException {
         final int maxDelay = 500;
 
         final BlockingQueue<StreamPayload> payload = new LinkedBlockingQueue<>();

providers/flagd/src/test/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/connector/grpc/GrpcStreamConnectorTest.java

@@ -1,18 +1,5 @@
 package dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.grpc;
 
-import dev.openfeature.contrib.providers.flagd.FlagdOptions;
-import dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.StreamPayload;
-import dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.StreamPayloadType;
-import dev.openfeature.flagd.sync.FlagSyncServiceGrpc;
-import dev.openfeature.flagd.sync.SyncService;
-import org.junit.jupiter.api.Test;
-import org.mockito.Mockito;
-
-import java.lang.reflect.Field;
-import java.time.Duration;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.TimeUnit;
-
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
@@ -21,6 +8,20 @@
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
+import java.lang.reflect.Field;
+import java.time.Duration;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import dev.openfeature.contrib.providers.flagd.FlagdOptions;
+import dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.StreamPayload;
+import dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.StreamPayloadType;
+import dev.openfeature.flagd.sync.FlagSyncServiceGrpc;
+import dev.openfeature.flagd.sync.SyncService;
+
 class GrpcStreamConnectorTest {
 
     private static final Duration MAX_WAIT_MS = Duration.ofMillis(500);