fix(flagd): improve stream observer, refine retry policy; don't use retry to avoid busy loop

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

@@ -14,6 +14,7 @@ public final class Config {
     static final String DEFAULT_HOST = "localhost";
 
     static final int DEFAULT_DEADLINE = 500;
+    static final int DEFAULT_MAX_RETRY_BACKOFF = 5;
     static final int DEFAULT_STREAM_DEADLINE_MS = 10 * 60 * 1000;
     static final int DEFAULT_STREAM_RETRY_GRACE_PERIOD = 5;
     static final int DEFAULT_MAX_CACHE_SIZE = 1000;
@@ -31,6 +32,7 @@ public final class Config {
     static final String MAX_CACHE_SIZE_ENV_VAR_NAME = "FLAGD_MAX_CACHE_SIZE";
     static final String BASE_EVENT_STREAM_RETRY_BACKOFF_MS_ENV_VAR_NAME = "FLAGD_RETRY_BACKOFF_MS";
     static final String DEADLINE_MS_ENV_VAR_NAME = "FLAGD_DEADLINE_MS";
+    static final String FLAGD_RETRY_BACKOFF_MAX_MS_VAR_NAME = "FLAGD_RETRY_BACKOFF_MAX_MS";
     static final String STREAM_DEADLINE_MS_ENV_VAR_NAME = "FLAGD_STREAM_DEADLINE_MS";
     static final String SOURCE_SELECTOR_ENV_VAR_NAME = "FLAGD_SOURCE_SELECTOR";
     /**

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

@@ -98,10 +98,18 @@ public class FlagdOptions {
     @Builder.Default
     private int deadline = fallBackToEnvOrDefault(Config.DEADLINE_MS_ENV_VAR_NAME, Config.DEFAULT_DEADLINE);
 
+    /**
+     * Max stream retry backoff in milliseconds.
+     */
+    @Builder.Default
+    private int retryBackoffMaxMs =
+            fallBackToEnvOrDefault(Config.FLAGD_RETRY_BACKOFF_MAX_MS_VAR_NAME, Config.DEFAULT_MAX_RETRY_BACKOFF);
+
     /**
      * Streaming connection deadline in milliseconds.
      * Set to 0 to disable the deadline.
-     * Defaults to 600000 (10 minutes); recommended to prevent infrastructure from killing idle connections.
+     * Defaults to 600000 (10 minutes); recommended to prevent infrastructure from
+     * killing idle connections.
      */
     @Builder.Default
     private int streamDeadlineMs =

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

@@ -24,67 +24,57 @@
 import javax.net.ssl.SSLException;
 
 /** gRPC channel builder helper. */
+@SuppressFBWarnings(value = "SE_BAD_FIELD", justification = "we don't care to serialize this")
 public class ChannelBuilder {
-
     /**
      * Controls retry (not-reconnection) policy for failed RPCs.
      */
     @SuppressWarnings({"unchecked", "rawtypes"})
-    static final Map<String, ?> SERVICE_CONFIG_WITH_RETRY = new HashMap() {
-        {
-            put("methodConfig", Arrays.asList(new HashMap() {
-                {
-                    put(
-                            "name",
-                            Arrays.asList(
-                                    new HashMap() {
-                                        {
-                                            put("service", "flagd.sync.v1.FlagSyncService");
-                                        }
-                                    },
-                                    new HashMap() {
-                                        {
-                                            put("service", "flagd.evaluation.v1.Service");
-                                        }
-                                    }));
-                    put("retryPolicy", new HashMap() {
-                        {
-                            // 1 + 2 + 4
-                            put("maxAttempts", 3.0); // types used here are important, need to be doubles
-                            put("initialBackoff", "1s");
-                            put("maxBackoff", "5s");
-                            put("backoffMultiplier", 2.0);
-                            // status codes to retry on:
-                            put(
-                                    "retryableStatusCodes",
-                                    Arrays.asList(
-                                            /*
-                                             * All codes are retryable except OK and DEADLINE_EXCEEDED since
-                                             * any others not listed here cause a very tight loop of retries.
-                                             * DEADLINE_EXCEEDED is typically a result of a client specified deadline,
-                                             * and definitionally should not result in a tight loop (it's a timeout).
-                                             */
-                                            Code.CANCELLED.toString(),
-                                            Code.UNKNOWN.toString(),
-                                            Code.INVALID_ARGUMENT.toString(),
-                                            Code.NOT_FOUND.toString(),
-                                            Code.ALREADY_EXISTS.toString(),
-                                            Code.PERMISSION_DENIED.toString(),
-                                            Code.RESOURCE_EXHAUSTED.toString(),
-                                            Code.FAILED_PRECONDITION.toString(),
-                                            Code.ABORTED.toString(),
-                                            Code.OUT_OF_RANGE.toString(),
-                                            Code.UNIMPLEMENTED.toString(),
-                                            Code.INTERNAL.toString(),
-                                            Code.UNAVAILABLE.toString(),
-                                            Code.DATA_LOSS.toString(),
-                                            Code.UNAUTHENTICATED.toString()));
-                        }
-                    });
-                }
-            }));
-        }
-    };
+    static Map<String, ?> buildRetryPolicy(final FlagdOptions options) {
+        return new HashMap() {
+            {
+                put("methodConfig", Arrays.asList(new HashMap() {
+                    {
+                        put(
+                                "name",
+                                Arrays.asList(
+                                        new HashMap() {
+                                            {
+                                                put("service", "flagd.sync.v1.FlagSyncService");
+                                            }
+                                        },
+                                        new HashMap() {
+                                            {
+                                                put("service", "flagd.evaluation.v1.Service");
+                                            }
+                                        }));
+                        put("retryPolicy", new HashMap() {
+                            {
+                                // 1 + 2 + 4
+                                put("maxAttempts", 3.0); // types used here are important, need to be doubles
+                                put("initialBackoff", "1s");
+                                put(
+                                        "maxBackoff",
+                                        options.getRetryBackoffMaxMs() >= 1000
+                                                ? String.format("%ds", options.getRetryBackoffMaxMs() / 1000)
+                                                : "1s");
+                                put("backoffMultiplier", 2.0);
+                                // status codes to retry on:
+                                put(
+                                        "retryableStatusCodes",
+                                        Arrays.asList(
+                                                /*
+                                                 * As per gRPC spec, the following status codes are safe to retry:
+                                                 * UNAVAILABLE, UNKNOWN,
+                                                 */
+                                                Code.UNKNOWN.toString(), Code.UNAVAILABLE.toString()));
+                            }
+                        });
+                    }
+                }));
+            }
+        };
+    }
 
     private ChannelBuilder() {}
 
@@ -109,7 +99,7 @@ public static ManagedChannel nettyChannel(final FlagdOptions options) {
                     .eventLoopGroup(new MultiThreadIoEventLoopGroup(EpollIoHandler.newFactory()))
                     .channelType(EpollDomainSocketChannel.class)
                     .usePlaintext()
-                    .defaultServiceConfig(SERVICE_CONFIG_WITH_RETRY)
+                    .defaultServiceConfig(buildRetryPolicy(options))
                     .enableRetry()
                     .build();
         }
@@ -155,7 +145,7 @@ public static ManagedChannel nettyChannel(final FlagdOptions options) {
                 builder.intercept(new FlagdGrpcInterceptor(options.getOpenTelemetry()));
             }
 
-            return builder.defaultServiceConfig(SERVICE_CONFIG_WITH_RETRY)
+            return builder.defaultServiceConfig(buildRetryPolicy(options))
                     .enableRetry()
                     .build();
         } catch (SSLException ssle) {

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

@@ -5,8 +5,6 @@
 import dev.openfeature.sdk.ProviderEvent;
 import io.grpc.ConnectivityState;
 import io.grpc.ManagedChannel;
-import io.grpc.stub.AbstractBlockingStub;
-import io.grpc.stub.AbstractStub;
 import java.util.Collections;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
@@ -16,12 +14,9 @@
 /**
  * A generic GRPC connector that manages connection states, reconnection logic, and event streaming for
  * GRPC services.
- *
- * @param <T> the type of the asynchronous stub for the GRPC service
- * @param <K> the type of the blocking stub for the GRPC service
  */
 @Slf4j
-public class ChannelConnector<T extends AbstractStub<T>, K extends AbstractBlockingStub<K>> {
+public class ChannelConnector {
 
     /**
      * The GRPC managed channel for managing the underlying GRPC connection.
@@ -48,22 +43,11 @@ public class ChannelConnector<T extends AbstractStub<T>, K extends AbstractBlock
      */
     public ChannelConnector(
             final FlagdOptions options, final Consumer<FlagdProviderEvent> onConnectionEvent, ManagedChannel channel) {
-
         this.channel = channel;
         this.deadline = options.getDeadline();
         this.onConnectionEvent = onConnectionEvent;
     }
 
-    /**
-     * Constructs a {@code ChannelConnector} instance for testing purposes.
-     *
-     * @param options             the configuration options for the GRPC connection
-     * @param onConnectionEvent   a consumer to handle connection events
-     */
-    public ChannelConnector(final FlagdOptions options, final Consumer<FlagdProviderEvent> onConnectionEvent) {
-        this(options, onConnectionEvent, ChannelBuilder.nettyChannel(options));
-    }
-
     /**
      * Initializes the GRPC connection by waiting for the channel to be ready and monitoring its state.
      *