perf(flagd): run all 3 e2e resolver modes concurrently via @TestFactory

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

@@ -8,23 +8,25 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.atomic.AtomicBoolean;
 import lombok.extern.slf4j.Slf4j;
+import org.testcontainers.DockerClientFactory;
 
 /**
  * A pool of pre-warmed {@link ContainerEntry} instances.
  *
- * <p>All containers are started in parallel during {@link #initialize()}, paying the ~45s Docker
- * Compose startup cost only once. Scenarios borrow a container via {@link #acquire()} and return
- * it via {@link #release(ContainerEntry)} after teardown, allowing the next scenario to reuse it
- * immediately without any cold-start overhead.
+ * <p>All containers are started in parallel on the first {@link #acquire()} call, paying the
+ * Docker Compose startup cost only once per JVM. Scenarios borrow a container via
+ * {@link #acquire()} and return it via {@link #release(ContainerEntry)} after teardown.
  *
- * <p>Pool size is controlled by the system property {@code flagd.e2e.pool.size} (default: 2).
+ * <p>Cleanup is handled automatically via a JVM shutdown hook — no explicit lifecycle calls are
+ * needed from test classes. This means multiple test classes (e.g. several {@code @Suite} runners
+ * or {@code @TestFactory} methods) share the same pool across the entire JVM lifetime without
+ * redundant container startups.
  *
- * <p>Multiple test classes may share the same JVM fork (Surefire {@code reuseForks=true}). Each
- * class calls {@link #initialize()} and {@link #shutdown()} once. A reference counter ensures
- * that containers are only started on the first {@code initialize()} call and only stopped when
- * the last {@code shutdown()} call is made, preventing one class from destroying containers that
- * are still in use by another class running concurrently in the same JVM.
+ * <p>Pool size is controlled by the system property {@code flagd.e2e.pool.size}
+ * (default: min(availableProcessors, 4)).
  */
 @Slf4j
 public class ContainerPool {
@@ -34,62 +36,128 @@ public class ContainerPool {
 
     private static final BlockingQueue<ContainerEntry> pool = new LinkedBlockingQueue<>();
     private static final List<ContainerEntry> all = new ArrayList<>();
-    private static final java.util.concurrent.atomic.AtomicInteger refCount =
-            new java.util.concurrent.atomic.AtomicInteger(0);
+    private static final AtomicBoolean initialized = new AtomicBoolean(false);
 
-    public static void initialize() throws Exception {
-        if (refCount.getAndIncrement() > 0) {
-            log.info("Container pool already initialized (refCount={}), reusing existing pool.", refCount.get());
+    /**
+     * JVM-wide semaphore that serializes disruptive container operations (stop/restart) across all
+     * parallel Cucumber engines. Only one scenario at a time may bring a container down, preventing
+     * cascading initialization timeouts in sibling scenarios that are waiting for a container slot.
+     */
+    private static final Semaphore restartSlot = new Semaphore(1);
+
+    static {
+        Runtime.getRuntime().addShutdownHook(new Thread(ContainerPool::stopAll, "container-pool-shutdown"));
+    }
+
+    /**
+     * Borrow a container from the pool, blocking until one becomes available.
+     * Initializes the pool on the first call. The caller MUST call
+     * {@link #release(ContainerEntry)} when done.
+     */
+    public static ContainerEntry acquire() throws Exception {
+        ensureInitialized();
+        return pool.take();
+    }
+
+    /** Return a container to the pool so the next scenario can use it. */
+    public static void release(ContainerEntry entry) {
+        pool.add(entry);
+    }
+
+    /**
+     * Acquires the JVM-wide restart slot before stopping or restarting a container.
+     * Must be paired with {@link #releaseRestartSlot()} in the scenario {@code @After} hook.
+     */
+    public static void acquireRestartSlot() throws InterruptedException {
+        log.debug("Acquiring restart slot...");
+        restartSlot.acquire();
+        log.debug("Restart slot acquired.");
+    }
+
+    /** Releases the JVM-wide restart slot acquired by {@link #acquireRestartSlot()}. */
+    public static void releaseRestartSlot() {
+        restartSlot.release();
+        log.debug("Restart slot released.");
+    }
+
+    private static void ensureInitialized() throws Exception {
+        if (initialized.get()) {
             return;
         }
-        log.info("Starting container pool of size {}...", POOL_SIZE);
-        ExecutorService executor = Executors.newFixedThreadPool(POOL_SIZE);
-        List<Future<ContainerEntry>> futures = new ArrayList<>();
-        try {
-            for (int i = 0; i < POOL_SIZE; i++) {
-                futures.add(executor.submit(ContainerEntry::start));
+        synchronized (ContainerPool.class) {
+            if (initialized.get()) {
+                return;
             }
-            for (Future<ContainerEntry> future : futures) {
-                ContainerEntry entry = future.get();
-                pool.add(entry);
-                all.add(entry);
-            }
-        } catch (Exception e) {
-            // Collect any containers that started successfully (including from in-flight futures)
-            for (Future<ContainerEntry> future : futures) {
-                try {
+            log.info("Starting container pool of size {}...", POOL_SIZE);
+            ExecutorService executor = null;
+            List<Future<ContainerEntry>> futures = new ArrayList<>();
+            try {
+                // Resolve the Testcontainers Docker client once, single-threaded, before starting
+                // containers in parallel. Its DockerClientProviderStrategy is loaded via a non
+                // thread-safe ServiceLoader; concurrent first-time resolution throws
+                // ConcurrentModificationException, leaving containers unstarted.
+                DockerClientFactory.instance().client();
+                executor = Executors.newFixedThreadPool(POOL_SIZE);
+                for (int i = 0; i < POOL_SIZE; i++) {
+                    futures.add(executor.submit(ContainerEntry::start));
+                }
+                for (Future<ContainerEntry> future : futures) {
                     ContainerEntry entry = future.get();
-                    if (!all.contains(entry)) {
-                        all.add(entry);
-                    }
-                } catch (Exception ignored) {
-                    // future failed; nothing to stop for it
+                    pool.add(entry);
+                    all.add(entry);
+                }
+                // Publish only after every container is up, so a failed startup leaves
+                // initialized=false and the next acquire() retries instead of blocking on an
+                // empty pool forever.
+                initialized.set(true);
+            } catch (Exception e) {
+                stopInFlight(futures, e);
+                pool.clear();
+                all.clear();
+                throw e;
+            } finally {
+                if (executor != null) {
+                    // shutdownNow interrupts container starts still running after a failure.
+                    executor.shutdownNow();
                 }
             }
-            all.forEach(entry -> {
-                try {
-                    entry.stop();
-                } catch (IOException suppressed) {
-                    e.addSuppressed(suppressed);
+            log.info("Container pool ready ({} containers).", POOL_SIZE);
+        }
+    }
+
+    /**
+     * Stop every container produced during a failed startup — both those already recorded in
+     * {@link #all} and any that an in-flight future completed after the failure — so a leaked
+     * container can't race the next initialization attempt. Cancels futures still running.
+     */
+    private static void stopInFlight(List<Future<ContainerEntry>> futures, Exception failure) {
+        all.forEach(entry -> stopQuietly(entry, failure));
+        for (Future<ContainerEntry> future : futures) {
+            if (future.cancel(true)) {
+                continue; // wasn't started yet — nothing to stop
+            }
+            try {
+                ContainerEntry entry = future.get();
+                if (!all.contains(entry)) {
+                    stopQuietly(entry, failure);
                 }
-            });
-            pool.clear();
-            all.clear();
-            refCount.decrementAndGet();
-            throw e;
-        } finally {
-            executor.shutdown();
+            } catch (Exception ignored) {
+                // future failed or was interrupted; no container to stop
+            }
         }
-        log.info("Container pool ready ({} containers).", POOL_SIZE);
     }
 
-    public static void shutdown() {
-        int remaining = refCount.decrementAndGet();
-        if (remaining > 0) {
-            log.info("Container pool still in use by {} class(es), deferring shutdown.", remaining);
-            return;
+    private static void stopQuietly(ContainerEntry entry, Exception failure) {
+        try {
+            entry.stop();
+        } catch (IOException suppressed) {
+            failure.addSuppressed(suppressed);
         }
-        log.info("Last shutdown call — stopping all containers.");
+    }
+
+    private static void stopAll() {
+        if (all.isEmpty()) return;
+        log.info("Shutdown hook — stopping all containers.");
         all.forEach(entry -> {
             try {
                 entry.stop();
@@ -100,17 +168,4 @@ public static void shutdown() {
         pool.clear();
         all.clear();
     }
-
-    /**
-     * Borrow a container from the pool, blocking until one becomes available.
-     * The caller MUST call {@link #release(ContainerEntry)} when done.
-     */
-    public static ContainerEntry acquire() throws InterruptedException {
-        return pool.take();
-    }
-
-    /** Return a container to the pool so the next scenario can use it. */
-    public static void release(ContainerEntry entry) {
-        pool.add(entry);
-    }
 }

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

@@ -1,6 +1,9 @@
 package dev.openfeature.contrib.providers.flagd.e2e;
 
 import dev.openfeature.contrib.providers.flagd.Config;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.Socket;
 import java.util.Optional;
 import org.testcontainers.containers.ComposeContainer;
 import org.testcontainers.containers.ContainerState;
@@ -29,4 +32,43 @@ public static String getLaunchpadUrl(ComposeContainer container) {
                 })
                 .orElseThrow(() -> new RuntimeException("Could not find launchpad url"));
     }
+
+    /**
+     * Blocks until the given flagd service port accepts TCP connections, or the timeout elapses.
+     * The launchpad's {@code /start} endpoint polls flagd's HTTP {@code /readyz} before returning,
+     * but the gRPC ports (8013, 8015) may become available slightly later. Waiting here prevents
+     * {@code setProviderAndWait} from timing out under parallel load.
+     */
+    public static void waitForGrpcPort(ComposeContainer container, Config.Resolver resolver, long timeoutMs)
+            throws InterruptedException {
+        int internalPort;
+        switch (resolver) {
+            case RPC:
+                internalPort = 8013;
+                break;
+            case IN_PROCESS:
+                internalPort = 8015;
+                break;
+            default:
+                return;
+        }
+        ContainerState state = container
+                .getContainerByServiceName("flagd")
+                .orElseThrow(() -> new RuntimeException("Could not find flagd container"));
+        String host = state.getHost();
+        int mappedPort = state.getMappedPort(internalPort);
+        long deadline = System.currentTimeMillis() + timeoutMs;
+        while (System.currentTimeMillis() < deadline) {
+            try (Socket s = new Socket()) {
+                s.connect(new InetSocketAddress(host, mappedPort), 100);
+                return;
+            } catch (IOException ignored) {
+                Thread.sleep(50);
+            }
+        }
+        // Surface the readiness failure here rather than letting setupProvider build a provider
+        // against a closed port and fail later with an opaque setProviderAndWait timeout.
+        throw new IllegalStateException(
+                "flagd gRPC port " + mappedPort + " did not accept connections within " + timeoutMs + "ms");
+    }
 }

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

@@ -0,0 +1,103 @@
+package dev.openfeature.contrib.providers.flagd.e2e;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import lombok.extern.slf4j.Slf4j;
+import org.junit.platform.engine.TestExecutionResult;
+import org.junit.platform.engine.reporting.ReportEntry;
+import org.junit.platform.launcher.TestExecutionListener;
+import org.junit.platform.launcher.TestIdentifier;
+import org.junit.platform.launcher.TestPlan;
+
+/**
+ * Captures the full lifecycle of a JUnit Platform test execution, tracking start, finish, and skip
+ * events for every node in the test plan (both containers and tests). Results are later replayed as
+ * JUnit Jupiter {@link org.junit.jupiter.api.DynamicTest} instances to expose the Cucumber scenario
+ * tree in IDEs.
+ */
+@Slf4j
+class CucumberResultListener implements TestExecutionListener {
+
+    private final Set<String> started = Collections.newSetFromMap(new ConcurrentHashMap<>());
+    private final Map<String, TestExecutionResult> results = new ConcurrentHashMap<>();
+    private final Map<String, String> skipped = new ConcurrentHashMap<>();
+
+    @Override
+    public void testPlanExecutionStarted(TestPlan testPlan) {
+        log.debug("Cucumber execution started");
+    }
+
+    @Override
+    public void testPlanExecutionFinished(TestPlan testPlan) {
+        log.debug(
+                "Cucumber execution finished — started={}, finished={}, skipped={}",
+                started.size(),
+                results.size(),
+                skipped.size());
+    }
+
+    @Override
+    public void executionStarted(TestIdentifier id) {
+        log.debug("  START  {}", id.getDisplayName());
+        started.add(id.getUniqueId());
+    }
+
+    @Override
+    public void executionFinished(TestIdentifier id, TestExecutionResult result) {
+        results.put(id.getUniqueId(), result);
+        if (result.getStatus() == TestExecutionResult.Status.FAILED) {
+            log.debug(
+                    "  FAIL   {} — {}",
+                    id.getDisplayName(),
+                    result.getThrowable().map(Throwable::getMessage).orElse("(no message)"));
+        } else {
+            log.debug("  {}   {}", result.getStatus(), id.getDisplayName());
+        }
+    }
+
+    @Override
+    public void executionSkipped(TestIdentifier id, String reason) {
+        skipped.put(id.getUniqueId(), reason);
+        log.debug("  SKIP   {} — {}", id.getDisplayName(), reason);
+    }
+
+    @Override
+    public void dynamicTestRegistered(TestIdentifier id) {
+        log.debug("  DYN    {}", id.getDisplayName());
+    }
+
+    @Override
+    public void reportingEntryPublished(TestIdentifier id, ReportEntry entry) {
+        log.debug("  REPORT {} — {}", id.getDisplayName(), entry);
+    }
+
+    /** Whether the node with the given unique ID had {@code executionStarted} called. */
+    boolean wasStarted(String uniqueId) {
+        return started.contains(uniqueId);
+    }
+
+    /** Whether the node was skipped before starting. */
+    boolean wasSkipped(String uniqueId) {
+        return skipped.containsKey(uniqueId);
+    }
+
+    /** The skip reason for a skipped node, or {@code null} if not skipped. */
+    String getSkipReason(String uniqueId) {
+        return skipped.get(uniqueId);
+    }
+
+    /** Whether a finished result was recorded for the given node. */
+    boolean hasResult(String uniqueId) {
+        return results.containsKey(uniqueId);
+    }
+
+    /**
+     * The recorded {@link TestExecutionResult}, or {@code null} if the node never finished.
+     * Use {@link #hasResult} to distinguish "finished with success" from "never finished".
+     */
+    TestExecutionResult getResult(String uniqueId) {
+        return results.get(uniqueId);
+    }
+}