fatal only on first connection

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

Author: leakonvalinka

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

@@ -38,6 +38,7 @@ public class SyncStreamQueueSource implements QueueSource {
 
     private final AtomicBoolean shutdown = new AtomicBoolean(false);
     private final AtomicBoolean shouldThrottle = new AtomicBoolean(false);
+    private final AtomicBoolean successfulSync = new AtomicBoolean(false);
     private final int streamDeadline;
     private final int deadline;
     private final int maxBackoffMs;
@@ -137,7 +138,7 @@ private void observeSyncStream() {
                 try {
                     observer.metadata = getMetadata();
                 } catch (StatusRuntimeException metaEx) {
-                    if (fatalStatusCodes.contains(metaEx.getStatus().getCode().name())) {
+                    if (fatalStatusCodes.contains(metaEx.getStatus().getCode().name()) && !successfulSync.get()) {
                         log.debug("Fatal status code for metadata request: {}, not retrying", metaEx.getStatus().getCode());
                         enqueueFatal(String.format("Fatal: Failed to connect for metadata request, not retrying for error %s", metaEx.getStatus().getCode()));
                     } else {
@@ -152,8 +153,9 @@ private void observeSyncStream() {
 
                 try {
                     syncFlags(observer);
+                    successfulSync.set(true);
                 } catch (StatusRuntimeException ex) {
-                    if (fatalStatusCodes.contains(ex.getStatus().getCode().toString())) {
+                    if (fatalStatusCodes.contains(ex.getStatus().getCode().toString()) && !successfulSync.get()) {
                         log.debug("Fatal status code during sync stream: {}, not retrying", ex.getStatus().getCode());
                         enqueueFatal(String.format("Fatal: Failed to connect for metadata request, not retrying for error %s", ex.getStatus().getCode()));
                     } else {

providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/rpc/RpcResolver.java

@@ -60,6 +60,7 @@
 public final class RpcResolver implements Resolver {
     private static final int QUEUE_SIZE = 5;
     private final AtomicBoolean shutdown = new AtomicBoolean(false);
+    private final AtomicBoolean successfulConnection = new AtomicBoolean(false);
     private final ChannelConnector connector;
     private final Cache cache;
     private final ResolveStrategy strategy;
@@ -351,18 +352,20 @@ private void observeEventStream() throws InterruptedException {
 
                 Throwable streamException = taken.getError();
                 if (streamException != null) {
-                    log.debug(
-                            "Exception in event stream connection, streamException {}, will reconnect",
-                            streamException);
                     if (streamException instanceof StatusRuntimeException && fatalStatusCodes.contains(
-                            ((StatusRuntimeException) streamException).getStatus().getCode().name())) {
+                            ((StatusRuntimeException) streamException).getStatus().getCode().name()) && !successfulConnection.get()) {
+                        log.debug("Fatal error code received: {}", ((StatusRuntimeException) streamException).getStatus().getCode());
                         this.handleFatalError();
                     } else {
+                        log.debug(
+                                "Exception in event stream connection, streamException {}, will reconnect",
+                                streamException);
                         this.handleErrorOrComplete();
                     }
                     break;
                 }
 
+                successfulConnection.set(true);
                 final EventStreamResponse response = taken.getResponse();
                 log.debug("Got stream response: {}", response);