add shutdown after fatal, fix tests

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

Author: leakonvalinka

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

@@ -125,7 +125,7 @@ public class FlagdOptions {
             fallBackToEnvOrDefault(Config.STREAM_RETRY_GRACE_PERIOD, Config.DEFAULT_STREAM_RETRY_GRACE_PERIOD);
 
     /**
-     * List of grpc response status codes for which failed connections are not retried.
+     * List of grpc response status codes for which the provider transitions into fatal state upon first connection.
      * Defaults to empty list
      */
     @Builder.Default

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

@@ -240,6 +240,7 @@ private void onProviderEvent(FlagdProviderEvent flagdProviderEvent) {
 
     private void onError() {
         this.emitProviderError(ProviderEventDetails.builder().errorCode(ErrorCode.PROVIDER_FATAL).build());
+        this.shutdown();
     }
 
     private void onConfigurationChanged(FlagdProviderEvent flagdProviderEvent) {

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

@@ -141,6 +141,7 @@ private void observeSyncStream() {
                     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()));
+                        return;
                     } else {
                         // retry for other status codes
                         String message = metaEx.getMessage();
@@ -158,6 +159,7 @@ private void observeSyncStream() {
                     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()));
+                        return;
                     } else {
                         // retry for other status codes
                         log.error("Unexpected sync stream exception, will restart.", ex);

providers/flagd/src/test/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/FlagStoreTest.java

@@ -64,14 +64,14 @@ void connectorHandling() throws Exception {
         });
 
         assertTimeoutPreemptively(Duration.ofMillis(maxDelay), () -> {
-            assertEquals(StorageState.ERROR, states.take().getStorageState());
+            assertEquals(StorageState.STALE, states.take().getStorageState());
         });
 
         // Shutdown handling
         store.shutdown();
 
         assertTimeoutPreemptively(Duration.ofMillis(maxDelay), () -> {
-            assertEquals(StorageState.ERROR, states.take().getStorageState());
+            assertEquals(StorageState.STALE, states.take().getStorageState());
         });
     }
 

providers/flagd/src/test/java/dev/openfeature/contrib/providers/flagd/resolver/rpc/RpcResolverTest.java

@@ -111,7 +111,7 @@ void onCompletedRerunsStreamWithError() throws Exception {
 
         // should run consumer with error
         await().untilAsserted(() ->
-                verify(consumer).accept(argThat((arg) -> arg.getEvent() == ProviderEvent.PROVIDER_ERROR)));
+                verify(consumer).accept(argThat((arg) -> arg.getEvent() == ProviderEvent.PROVIDER_STALE)));
         // should have restarted the stream (2 calls)
         await().untilAsserted(() -> verify(stub, times(2)).eventStream(any(), any()));
     }
@@ -128,7 +128,7 @@ void onErrorRunsConsumerWithError() throws Exception {
 
         // should run consumer with error
         await().untilAsserted(() ->
-                verify(consumer).accept(argThat((arg) -> arg.getEvent() == ProviderEvent.PROVIDER_ERROR)));
+                verify(consumer).accept(argThat((arg) -> arg.getEvent() == ProviderEvent.PROVIDER_STALE)));
         // should have restarted the stream (2 calls)
         await().untilAsserted(() -> verify(stub, times(2)).eventStream(any(), any()));
     }