fixup: fixing nit, and adding information for fallthrough

Signed-off-by: Simon Schrottner <[email protected]>

Author: aepfli

Diff for: providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/FlagdProvider.java

@@ -51,7 +51,7 @@ public class FlagdProvider extends EventProvider {
     /**
      * A scheduled task for emitting {@link ProviderEvent#PROVIDER_ERROR}.
      */
-    private ScheduledFuture<?> reconnectTask;
+    private ScheduledFuture<?> errorTask;
 
     /**
      * The grace period in milliseconds to wait after {@link ProviderEvent#PROVIDER_STALE} before emitting a
@@ -188,7 +188,7 @@ EvaluationContext getEnrichedContext() {
     }
 
     @SuppressWarnings("checkstyle:fallthrough")
-    private void onProviderEvent(FlagdProviderEvent flagdProviderEvent) {
+    private synchronized void onProviderEvent(FlagdProviderEvent flagdProviderEvent) {
 
         syncMetadata = flagdProviderEvent.getSyncMetadata();
         if (flagdProviderEvent.getSyncMetadata() != null) {
@@ -236,8 +236,8 @@ private void onReady() {
             initialized = true;
             log.info("initialized FlagdProvider");
         }
-        if (reconnectTask != null && !reconnectTask.isCancelled()) {
-            reconnectTask.cancel(false);
+        if (errorTask != null && !errorTask.isCancelled()) {
+            errorTask.cancel(false);
             log.debug("Reconnection task cancelled as connection became READY.");
         }
         this.emitProviderReady(
@@ -251,19 +251,22 @@ private void onError() {
                 .message("there has been an error")
                 .build());
 
-        if (reconnectTask != null && !reconnectTask.isCancelled()) {
-            reconnectTask.cancel(false);
+        if (errorTask != null && !errorTask.isCancelled()) {
+            errorTask.cancel(false);
         }
 
         if (!errorExecutor.isShutdown()) {
-            reconnectTask = errorExecutor.schedule(
+            errorTask = errorExecutor.schedule(
                     () -> {
-                        log.debug(
-                                "Provider did not reconnect successfully within {}s. Emit ERROR event...", gracePeriod);
-                        flagResolver.onError();
-                        this.emitProviderError(ProviderEventDetails.builder()
-                                .message("there has been an error")
-                                .build());
+                        if(previousEvent == ProviderEvent.PROVIDER_ERROR) {
+                            log.debug(
+                                    "Provider did not reconnect successfully within {}s. Emit ERROR event...",
+                                    gracePeriod);
+                            flagResolver.onError();
+                            this.emitProviderError(ProviderEventDetails.builder()
+                                    .message("there has been an error")
+                                    .build());
+                        }
                     },
                     gracePeriod,
                     TimeUnit.SECONDS);

Diff for: providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/connector/grpc/GrpcStreamConnector.java

@@ -16,7 +16,6 @@
 import io.grpc.Context.CancellableContext;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
 import lombok.extern.slf4j.Slf4j;
@@ -121,6 +120,7 @@ void observeEventStream(final BlockingQueue<QueuePayload> writeTo, final AtomicB
                     metadataException = e;
                 }
 
+                log.info("stream");
                 while (!shutdown.get()) {
                     final GrpcResponseModel response = streamReceiver.take();
                     if (response.isComplete()) {