Merge branch 'main' into fix/flagd-infinite-connection-retries

Signed-off-by: lea konvalinka <lea.konvalinka@dynatrace.com>

Author: leakonvalinka

providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/Config.java

@@ -23,6 +23,7 @@ public final class Config {
     static final int DEFAULT_MAX_CACHE_SIZE = 1000;
     static final int DEFAULT_OFFLINE_POLL_MS = 5000;
     static final long DEFAULT_KEEP_ALIVE = 0;
+    static final String DEFAULT_REINITIALIZE_ON_ERROR = "false";
 
     static final String RESOLVER_ENV_VAR = "FLAGD_RESOLVER";
     static final String HOST_ENV_VAR_NAME = "FLAGD_HOST";
@@ -55,6 +56,7 @@ public final class Config {
     static final String KEEP_ALIVE_MS_ENV_VAR_NAME = "FLAGD_KEEP_ALIVE_TIME_MS";
     static final String TARGET_URI_ENV_VAR_NAME = "FLAGD_TARGET_URI";
     static final String STREAM_RETRY_GRACE_PERIOD = "FLAGD_RETRY_GRACE_PERIOD";
+    static final String REINITIALIZE_ON_ERROR_ENV_VAR_NAME = "FLAGD_REINITIALIZE_ON_ERROR";
 
     static final String RESOLVER_RPC = "rpc";
     static final String RESOLVER_IN_PROCESS = "in-process";

providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/FlagdOptions.java

@@ -213,6 +213,16 @@ public class FlagdOptions {
     @Builder.Default
     private String defaultAuthority = fallBackToEnvOrDefault(Config.DEFAULT_AUTHORITY_ENV_VAR_NAME, null);
 
+    /**
+     * !EXPERIMENTAL!
+     * Whether to reinitialize the channel (TCP connection) after the grace period is exceeded.
+     * This can help recover from connection issues by creating fresh connections.
+     * Particularly useful for troubleshooting network issues related to proxies or service meshes.
+     */
+    @Builder.Default
+    private boolean reinitializeOnError = Boolean.parseBoolean(
+            fallBackToEnvOrDefault(Config.REINITIALIZE_ON_ERROR_ENV_VAR_NAME, Config.DEFAULT_REINITIALIZE_ON_ERROR));
+
     /**
      * Builder overwrite in order to customize the "build" method.
      *

providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/FlagdProvider.java

@@ -92,7 +92,7 @@ public FlagdProvider(final FlagdOptions options) {
         }
         hooks.add(new SyncMetadataHook(this::getEnrichedContext));
         contextEnricher = options.getContextEnricher();
-        errorExecutor = Executors.newSingleThreadScheduledExecutor();
+        errorExecutor = Executors.newSingleThreadScheduledExecutor(new FlagdThreadFactory("flagd-provider-thread"));
         gracePeriod = options.getRetryGracePeriod();
         deadline = options.getDeadline();
     }
@@ -106,7 +106,7 @@ public FlagdProvider(final FlagdOptions options) {
         deadline = Config.DEFAULT_DEADLINE;
         gracePeriod = Config.DEFAULT_STREAM_RETRY_GRACE_PERIOD;
         hooks.add(new SyncMetadataHook(this::getEnrichedContext));
-        errorExecutor = Executors.newSingleThreadScheduledExecutor();
+        errorExecutor = Executors.newSingleThreadScheduledExecutor(new FlagdThreadFactory("flagd-provider-thread"));
         if (initialized) {
             this.syncResources.initialize();
         }

providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/FlagdThreadFactory.java

@@ -0,0 +1,31 @@
+package dev.openfeature.contrib.providers.flagd;
+
+import java.util.Objects;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Thread factory for the Flagd provider to allow named daemon threads to be created.
+ */
+class FlagdThreadFactory implements ThreadFactory {
+
+    private final AtomicInteger counter = new AtomicInteger();
+    private final String namePrefix;
+
+    /**
+     * {@link FlagdThreadFactory}'s constructor.
+     *
+     * @param namePrefix    Prefix used for setting the new thread's name.
+     */
+    FlagdThreadFactory(String namePrefix) {
+        this.namePrefix = Objects.requireNonNull(namePrefix, "namePrefix must not be null");
+    }
+
+    @Override
+    public Thread newThread(Runnable runnable) {
+        final Thread thread = new Thread(runnable);
+        thread.setDaemon(true);
+        thread.setName(namePrefix + "-" + counter.incrementAndGet());
+        return thread;
+    }
+}

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

@@ -41,6 +41,7 @@ public class InProcessResolver implements Resolver {
     private final Consumer<FlagdProviderEvent> onConnectionEvent;
     private final Operator operator;
     private final String scope;
+    private final QueueSource queueSource;
 
     /**
      * Resolves flag values using
@@ -52,7 +53,8 @@ public class InProcessResolver implements Resolver {
      *                          connection/stream
      */
     public InProcessResolver(FlagdOptions options, Consumer<FlagdProviderEvent> onConnectionEvent) {
-        this.flagStore = new FlagStore(getConnector(options));
+        this.queueSource = getQueueSource(options);
+        this.flagStore = new FlagStore(queueSource);
         this.onConnectionEvent = onConnectionEvent;
         this.operator = new Operator();
         this.scope = options.getSelector();
@@ -97,6 +99,19 @@ public void init() throws Exception {
         stateWatcher.start();
     }
 
+    /**
+     * Called when the provider enters error state after grace period.
+     * Attempts to reinitialize the sync connector if enabled.
+     */
+    @Override
+    public void onError() {
+        if (queueSource instanceof SyncStreamQueueSource) {
+            SyncStreamQueueSource syncConnector = (SyncStreamQueueSource) queueSource;
+            // only reinitialize if option is enabled
+            syncConnector.reinitializeChannelComponents();
+        }
+    }
+
     /**
      * Shutdown in-process resolver.
      *
@@ -150,7 +165,7 @@ public ProviderEvaluation<Value> objectEvaluation(String key, Value defaultValue
                 .build();
     }
 
-    static QueueSource getConnector(final FlagdOptions options) {
+    static QueueSource getQueueSource(final FlagdOptions options) {
         if (options.getCustomConnector() != null) {
             return options.getCustomConnector();
         }

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

@@ -27,12 +27,13 @@
 import lombok.extern.slf4j.Slf4j;
 
 /**
- * Implements the {@link QueueSource} contract and emit flags obtained from flagd sync gRPC contract.
+ * Implements the {@link QueueSource} contract and emit flags obtained from
+ * flagd sync gRPC contract.
  */
 @Slf4j
 @SuppressFBWarnings(
         value = {"EI_EXPOSE_REP"},
-        justification = "Random is used to generate a variation & flag configurations require exposing")
+        justification = "We need to expose the BlockingQueue to allow consumers to read from it")
 public class SyncStreamQueueSource implements QueueSource {
     private static final int QUEUE_SIZE = 5;
 
@@ -45,14 +46,32 @@ public class SyncStreamQueueSource implements QueueSource {
     private final String selector;
     private final String providerId;
     private final boolean syncMetadataDisabled;
-    private final ChannelConnector channelConnector;
+    private final boolean reinitializeOnError;
+    private final FlagdOptions options;
     private final BlockingQueue<QueuePayload> outgoingQueue = new LinkedBlockingQueue<>(QUEUE_SIZE);
-    private final FlagSyncServiceStub flagSyncStub;
-    private final FlagSyncServiceBlockingStub metadataStub;
     private final List<String> fatalStatusCodes;
+    private volatile GrpcComponents grpcComponents;
 
     /**
-     * Creates a new SyncStreamQueueSource responsible for observing the event stream.
+     * Container for gRPC components to ensure atomicity during reinitialization.
+     * All three components are updated together to prevent consumers from seeing
+     * an inconsistent state where components are from different channel instances.
+     */
+    private static class GrpcComponents {
+        final ChannelConnector channelConnector;
+        final FlagSyncServiceStub flagSyncStub;
+        final FlagSyncServiceBlockingStub metadataStub;
+
+        GrpcComponents(ChannelConnector connector, FlagSyncServiceStub stub, FlagSyncServiceBlockingStub blockingStub) {
+            this.channelConnector = connector;
+            this.flagSyncStub = stub;
+            this.metadataStub = blockingStub;
+        }
+    }
+
+    /**
+     * Creates a new SyncStreamQueueSource responsible for observing the event
+     * stream.
      */
     public SyncStreamQueueSource(final FlagdOptions options) {
         streamDeadline = options.getStreamDeadlineMs();
@@ -61,12 +80,10 @@ public SyncStreamQueueSource(final FlagdOptions options) {
         providerId = options.getProviderId();
         maxBackoffMs = options.getRetryBackoffMaxMs();
         syncMetadataDisabled = options.isSyncMetadataDisabled();
-        channelConnector = new ChannelConnector(options, ChannelBuilder.nettyChannel(options));
-        flagSyncStub =
-                FlagSyncServiceGrpc.newStub(channelConnector.getChannel()).withWaitForReady();
-        metadataStub = FlagSyncServiceGrpc.newBlockingStub(channelConnector.getChannel())
-                .withWaitForReady();
         fatalStatusCodes = options.getFatalStatusCodes();
+        reinitializeOnError = options.isReinitializeOnError();
+        this.options = options;
+        initializeChannelComponents();
     }
 
     // internal use only
@@ -79,12 +96,51 @@ protected SyncStreamQueueSource(
         deadline = options.getDeadline();
         selector = options.getSelector();
         providerId = options.getProviderId();
-        channelConnector = connectorMock;
         maxBackoffMs = options.getRetryBackoffMaxMs();
-        flagSyncStub = stubMock;
         syncMetadataDisabled = options.isSyncMetadataDisabled();
-        metadataStub = blockingStubMock;
         fatalStatusCodes = options.getFatalStatusCodes();
+        reinitializeOnError = options.isReinitializeOnError();
+        this.options = options;
+        this.grpcComponents = new GrpcComponents(connectorMock, stubMock, blockingStubMock);
+    }
+
+    /** Initialize channel connector and stubs. */
+    private synchronized void initializeChannelComponents() {
+        ChannelConnector newConnector = new ChannelConnector(options, ChannelBuilder.nettyChannel(options));
+        FlagSyncServiceStub newFlagSyncStub =
+                FlagSyncServiceGrpc.newStub(newConnector.getChannel()).withWaitForReady();
+        FlagSyncServiceBlockingStub newMetadataStub =
+                FlagSyncServiceGrpc.newBlockingStub(newConnector.getChannel()).withWaitForReady();
+
+        // atomic assignment of all components as a single unit
+        grpcComponents = new GrpcComponents(newConnector, newFlagSyncStub, newMetadataStub);
+    }
+
+    /** Reinitialize channel connector and stubs on error. */
+    public synchronized void reinitializeChannelComponents() {
+        if (!reinitializeOnError || shutdown.get()) {
+            return;
+        }
+
+        log.info("Reinitializing channel gRPC components in attempt to restore stream.");
+        GrpcComponents oldComponents = grpcComponents;
+
+        try {
+            // create new channel components first
+            initializeChannelComponents();
+        } catch (Exception e) {
+            log.error("Failed to reinitialize channel components", e);
+            return;
+        }
+
+        // shutdown old connector after successful reinitialization
+        if (oldComponents != null && oldComponents.channelConnector != null) {
+            try {
+                oldComponents.channelConnector.shutdown();
+            } catch (Exception e) {
+                log.debug("Error shutting down old channel connector during reinitialization", e);
+            }
+        }
     }
 
     /** Initialize sync stream connector. */
@@ -111,7 +167,7 @@ public void shutdown() throws InterruptedException {
             log.debug("Shutdown already in progress or completed");
             return;
         }
-        this.channelConnector.shutdown();
+        grpcComponents.channelConnector.shutdown();
     }
 
     /** Contains blocking calls, to be used concurrently. */
@@ -175,13 +231,14 @@ private void observeSyncStream() {
         log.info("Shutdown invoked, exiting event stream listener");
     }
 
-    // TODO: remove the metadata call entirely after https://github.com/open-feature/flagd/issues/1584
+    // TODO: remove the metadata call entirely after
+    // https://github.com/open-feature/flagd/issues/1584
     private Struct getMetadata() {
         if (syncMetadataDisabled) {
             return null;
         }
 
-        FlagSyncServiceBlockingStub localStub = metadataStub;
+        FlagSyncServiceBlockingStub localStub = grpcComponents.metadataStub;
 
         if (deadline > 0) {
             localStub = localStub.withDeadlineAfter(deadline, TimeUnit.MILLISECONDS);
@@ -196,7 +253,8 @@ private Struct getMetadata() {
 
             return null;
         } catch (StatusRuntimeException e) {
-            // In newer versions of flagd, metadata is part of the sync stream. If the method is unimplemented, we
+            // In newer versions of flagd, metadata is part of the sync stream. If the
+            // method is unimplemented, we
             // can ignore the error
             if (e.getStatus() != null
                     && Status.Code.UNIMPLEMENTED.equals(e.getStatus().getCode())) {
@@ -208,7 +266,7 @@ private Struct getMetadata() {
     }
 
     private void syncFlags(SyncStreamObserver streamObserver) {
-        FlagSyncServiceStub localStub = flagSyncStub; // don't mutate the stub
+        FlagSyncServiceStub localStub = grpcComponents.flagSyncStub; // don't mutate the stub
         if (streamDeadline > 0) {
             localStub = localStub.withDeadlineAfter(streamDeadline, TimeUnit.MILLISECONDS);
         }

providers/flagd/src/test/java/dev/openfeature/contrib/providers/flagd/FlagdProviderSyncResourcesTest.java

@@ -79,14 +79,14 @@ void interruptingWaitingThread_isIgnored() throws InterruptedException {
     @Test
     void callingInitialize_wakesUpWaitingThread() throws InterruptedException {
         final AtomicBoolean isWaiting = new AtomicBoolean();
-        final AtomicBoolean successfulTest = new AtomicBoolean();
+        final AtomicLong waitTime = new AtomicLong(Long.MAX_VALUE);
         Thread waitingThread = new Thread(() -> {
             long start = System.currentTimeMillis();
             isWaiting.set(true);
             flagdProviderSyncResources.waitForInitialization(10000);
             long end = System.currentTimeMillis();
             long duration = end - start;
-            successfulTest.set(duration < MAX_TIME_TOLERANCE * 2);
+            waitTime.set(duration);
         });
         waitingThread.start();
 
@@ -100,7 +100,11 @@ void callingInitialize_wakesUpWaitingThread() throws InterruptedException {
 
         waitingThread.join();
 
-        Assertions.assertTrue(successfulTest.get());
+        Assertions.assertTrue(
+                waitTime.get() < MAX_TIME_TOLERANCE * 2,
+                () -> "Wakeup should be almost instant, but took " + waitTime.get()
+                        + " ms, which is more than the max of"
+                        + (MAX_TIME_TOLERANCE * 2) + " ms");
     }
 
     @Timeout(2)

providers/flagd/src/test/java/dev/openfeature/contrib/providers/flagd/FlagdThreadFactoryTest.java

@@ -0,0 +1,46 @@
+package dev.openfeature.contrib.providers.flagd;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import org.junit.jupiter.api.Test;
+
+class FlagdThreadFactoryTest {
+
+    private static final String THREAD_NAME = "testthread";
+    private final Runnable runnable = () -> {};
+
+    @Test
+    void verifyThreadFactoryThrowsNullPointerExceptionWhenNamePrefixIsNull() {
+
+        // Then
+        var exception = assertThrows(NullPointerException.class, () -> {
+            // When
+            new FlagdThreadFactory(null);
+        });
+        assertThat(exception.toString()).contains("namePrefix must not be null");
+    }
+
+    @Test
+    void verifyNewThreadHasNamePrefix() {
+
+        var flagdThreadFactory = new FlagdThreadFactory(THREAD_NAME);
+        var thread = flagdThreadFactory.newThread(runnable);
+
+        assertThat(thread.getName()).isEqualTo(THREAD_NAME + "-1");
+        assertThat(thread.isDaemon()).isTrue();
+    }
+
+    @Test
+    void verifyNewThreadHasNamePrefixWithIncrement() {
+
+        var flagdThreadFactory = new FlagdThreadFactory(THREAD_NAME);
+        var threadOne = flagdThreadFactory.newThread(runnable);
+        var threadTwo = flagdThreadFactory.newThread(runnable);
+
+        assertThat(threadOne.getName()).isEqualTo(THREAD_NAME + "-1");
+        assertThat(threadOne.isDaemon()).isTrue();
+        assertThat(threadTwo.getName()).isEqualTo(THREAD_NAME + "-2");
+        assertThat(threadTwo.isDaemon()).isTrue();
+    }
+}

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

@@ -9,15 +9,14 @@
 import dev.openfeature.sdk.FeatureProvider;
 import dev.openfeature.sdk.FlagEvaluationDetails;
 import dev.openfeature.sdk.MutableContext;
-import java.util.LinkedList;
-import java.util.List;
 import java.util.Optional;
+import java.util.concurrent.ConcurrentLinkedQueue;
 
 public class State {
     public ProviderType providerType;
     public Client client;
     public FeatureProvider provider;
-    public List<Event> events = new LinkedList<>();
+    public ConcurrentLinkedQueue<Event> events = new ConcurrentLinkedQueue<>();
     public Optional<Event> lastEvent;
     public FlagSteps.Flag flag;
     public MutableContext context = new MutableContext();