chore: log tweaks, retry cancels, add options.toBuidler (open-feature#1276)

Signed-off-by: Todd Baert <[email protected]>

Author: toddbaert

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

@@ -19,7 +19,7 @@
 /**
  * FlagdOptions is a builder to build flagd provider options.
  */
-@Builder
+@Builder(toBuilder = true)
 @Getter
 @SuppressWarnings("PMD.TooManyStaticImports")
 public class FlagdOptions {

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

@@ -61,7 +61,7 @@ public void waitForInitialization(long deadline) {
             // if wait(0) is called, the thread would wait forever, so we abort when this would happen
             if (now >= end) {
                 throw new GeneralError(String.format(
-                        "Deadline exceeded. Condition did not complete within the %d ms deadline", deadline));
+                        "Initialization timeout exceeded; did not complete within the %d ms deadline.", deadline));
             }
             long remaining = end - now;
             synchronized (this) {

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

@@ -58,12 +58,12 @@ public class ChannelBuilder {
                                     "retryableStatusCodes",
                                     Arrays.asList(
                                             /*
-                                             * All codes are retryable except OK, CANCELLED and DEADLINE_EXCEEDED since
+                                             * All codes are retryable except OK and DEADLINE_EXCEEDED since
                                              * any others not listed here cause a very tight loop of retries.
-                                             * CANCELLED is not retryable because it is a client-side termination.
                                              * 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(),

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

@@ -70,7 +70,7 @@ public ChannelConnector(final FlagdOptions options, final Consumer<FlagdProvider
      * @throws Exception if the channel does not reach the desired state within the deadline
      */
     public void initialize() throws Exception {
-        log.info("Initializing GRPC connection...");
+        log.info("Initializing GRPC connection.");
         monitorChannelState(ConnectivityState.READY);
     }
 
@@ -80,7 +80,7 @@ public void initialize() throws Exception {
      * @throws InterruptedException if interrupted while waiting for termination
      */
     public void shutdown() throws InterruptedException {
-        log.info("Shutting down GRPC connection...");
+        log.info("Shutting down GRPC connection.");
 
         if (!channel.isShutdown()) {
             channel.shutdownNow();

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

@@ -17,6 +17,7 @@ public class QueueingStreamObserver<T> implements StreamObserver<T> {
 
     public QueueingStreamObserver(final BlockingQueue<StreamResponseModel<T>> queue) {
         blockingQueue = queue;
+        queue.clear();
     }
 
     @Override

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

@@ -128,7 +128,7 @@ private void observeSyncStream() throws InterruptedException {
             // create a context which exists to track and cancel the stream
             try (CancellableContext context = Context.current().withCancellation()) {
 
-                restart(); // start the stream within the context
+                restart(); // start the stream with the context
 
                 // TODO: remove the metadata call entirely after https://github.com/open-feature/flagd/issues/1584
                 if (!syncMetadataDisabled) {
@@ -150,16 +150,18 @@ private void observeSyncStream() throws InterruptedException {
                 while (!shutdown.get() && !Context.current().isCancelled()) {
                     final StreamResponseModel<SyncFlagsResponse> taken = incomingQueue.take();
                     if (taken.isComplete()) {
-                        log.debug("Sync stream completed, will reconnect");
+                        log.debug("Sync stream completed, will restart");
                         // The stream is complete, we still try to reconnect
                         break;
                     }
 
                     Throwable streamException = taken.getError();
                     if (streamException != null) {
-                        log.debug("Exception in GRPC connection, streamException {}, will reconnect", streamException);
+                        log.debug("Exception in stream RPC, streamException {}, will restart", streamException);
                         if (!outgoingQueue.offer(new QueuePayload(
-                                QueuePayloadType.ERROR, "Error from stream or metadata", metadataResponse))) {
+                                QueuePayloadType.ERROR,
+                                String.format("Error from stream: %s", streamException.getMessage()),
+                                metadataResponse))) {
                             log.error("Failed to convey ERROR status, queue is full");
                         }
                         break;

providers/flagd/src/test/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/connector/sync/SyncStreamQueueSourceTest.java

@@ -40,6 +40,7 @@ class SyncStreamQueueSourceTest {
     @BeforeEach
     public void init() throws Exception {
         blockingStub = mock(FlagSyncServiceBlockingStub.class);
+        when(blockingStub.withDeadlineAfter(anyLong(), any())).thenReturn(blockingStub);
         when(blockingStub.getMetadata(any())).thenReturn(GetMetadataResponse.getDefaultInstance());
 
         mockConnector = mock(ChannelConnector.class);