flaky tests attempt, changing it to stop flagd all the time, without stopping the container

Signed-off-by: Simon Schrottner <[email protected]>

Author: aepfli

providers/flagd/spec

@@ -117,7 +117,45 @@ public GrpcConnector(
      */
     public void initialize() throws Exception {
         log.info("Initializing GRPC connection...");
-        ChannelMonitor.monitorChannelState(ConnectivityState.READY, channel, this::onReady, this::onConnectionLost);
+
+        monitorChannelState(ConnectivityState.READY, channel, this::onReady, this::onConnectionLost);
+    }
+
+    /**
+     * Monitors the state of a gRPC channel and triggers the specified callbacks based on state changes.
+     *
+     * @param expectedState     the initial state to monitor.
+     * @param channel           the ManagedChannel to monitor.
+     * @param onConnectionReady callback invoked when the channel transitions to a READY state.
+     * @param onConnectionLost  callback invoked when the channel transitions to a FAILURE or SHUTDOWN state.
+     */
+    public void monitorChannelState(
+            ConnectivityState expectedState,
+            ManagedChannel channel,
+            Runnable onConnectionReady,
+            Runnable onConnectionLost) {
+        channel.notifyWhenStateChanged(expectedState, () -> {
+            ConnectivityState currentState = channel.getState(true);
+            log.debug("Channel state changed to: {}", currentState);
+            if (currentState == ConnectivityState.READY) {
+                if (onConnectionReady != null) {
+                    onConnectionReady.run();
+                } else {
+                    log.debug("onConnectionReady is null");
+                }
+            } else if (currentState == ConnectivityState.TRANSIENT_FAILURE
+                    || currentState == ConnectivityState.SHUTDOWN) {
+                if (onConnectionLost != null) {
+                    onConnectionLost.run();
+                } else {
+                    log.debug("onConnectionLost is null");
+                }
+            }
+            if (currentState != ConnectivityState.SHUTDOWN) {
+                // Re-register the state monitor to watch for the next state transition.
+                monitorChannelState(currentState, channel, onConnectionReady, onConnectionLost);
+            }
+        });
     }
 
     /**
@@ -139,7 +177,7 @@ public void shutdown() throws InterruptedException {
 
         if (!channel.isShutdown()) {
             channel.shutdownNow();
-            channel.awaitTermination(deadline, TimeUnit.MILLISECONDS);
+            channel.awaitTermination(500, TimeUnit.MILLISECONDS);
         }
     }
 

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

@@ -25,7 +25,7 @@ public EventSteps(State state) {
     @Given("a {} event handler")
     public void a_stale_event_handler(String eventType) {
         state.client.on(mapEventType(eventType), eventDetails -> {
-            log.info("event tracked for {} at {} ms ", eventType, System.currentTimeMillis()%100_000);
+            log.info("event tracked for {} at {} ms ", eventType, System.currentTimeMillis() % 100_000);
             state.events.add(new Event(eventType, eventDetails));
         });
     }
@@ -58,9 +58,8 @@ public void eventHandlerShouldBeExecuted(String eventType) {
     @Then("the {} event handler should have been executed within {int}ms")
     public void eventHandlerShouldBeExecutedWithin(String eventType, int ms) {
         log.info("waiting for eventtype: {}", eventType);
-        await().atMost(ms, MILLISECONDS)
-                .pollInterval(10, MILLISECONDS)
-                .until(() -> state.events.stream().anyMatch(event -> event.type.equals(eventType)));
+        await().atMost(ms, MILLISECONDS).pollInterval(10, MILLISECONDS).until(() -> state.events.stream()
+                .anyMatch(event -> event.type.equals(eventType)));
         state.lastEvent = state.events.stream()
                 .filter(event -> event.type.equals(eventType))
                 .findFirst();

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

@@ -25,7 +25,6 @@
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.RandomStringUtils;
 import org.junit.jupiter.api.parallel.Isolated;
-import org.junit.platform.suite.api.BeforeSuite;
 import org.testcontainers.containers.BindMode;
 import org.testcontainers.shaded.org.apache.commons.io.FileUtils;
 
@@ -42,7 +41,6 @@ public ProviderSteps(State state) {
         super(state);
     }
 
-    /*
     @BeforeAll
     public static void beforeAll() throws IOException {
         State.resolverType = Config.Resolver.RPC;
@@ -68,31 +66,9 @@ public void before() throws IOException {
     @After
     public void tearDown() {
         OpenFeatureAPI.getInstance().shutdown();
-    }
-    */
-
-    @BeforeAll
-    public static void beforeAll() {
-        State.resolverType = Config.Resolver.RPC;
-    }
-
-    @Before
-    public void before() throws IOException {
-        state.events.clear();
-        sharedTempDir = Files.createDirectories(
-                Paths.get("tmp/" + RandomStringUtils.randomAlphanumeric(8).toLowerCase() + "/"));
-        container = new FlagdContainer()
-                .withFileSystemBind(sharedTempDir.toAbsolutePath().toString(), "/tmp", BindMode.READ_WRITE);
-        if (!container.isRunning()) {
-            container.start();
-        }
-    }
-
-    @After
-    public void tearDown() throws IOException {
-        OpenFeatureAPI.getInstance().shutdown();
-        container.stop();
-        FileUtils.deleteDirectory(sharedTempDir.toFile());
+        when().post("http://" + container.getLaunchpadUrl() + "/stop")
+                .then()
+                .statusCode(200);
     }
 
     @Given("a {} flagd provider")
@@ -169,8 +145,11 @@ public void setupProvider(String providerType) throws IOException, InterruptedEx
 
     @When("the connection is lost for {int}s")
     public void the_connection_is_lost_for(int seconds) throws InterruptedException {
-        log.info("Timeout and wait for {} seconds starting at {} ms, should resume at {} ms", seconds,
-                System.currentTimeMillis() % 100_000, System.currentTimeMillis() % 100_000 + seconds * 1000L);
+        log.info(
+                "Timeout and wait for {} seconds starting at {} ms, should resume at {} ms",
+                seconds,
+                System.currentTimeMillis() % 100_000,
+                System.currentTimeMillis() % 100_000 + seconds * 1000L);
 
         when().post("http://" + container.getLaunchpadUrl() + "/restart?seconds={seconds}", seconds)
                 .then()