fix: thread memory leak in InProcessResolver

providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/process/InProcessResolver.java

@@ -28,6 +28,8 @@
 import dev.openfeature.sdk.exceptions.TypeMismatchError;
 import dev.openfeature.sdk.internal.TriConsumer;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 
@@ -43,6 +45,8 @@ public class InProcessResolver implements Resolver {
     private final Operator operator;
     private final String scope;
     private final QueueSource queueSource;
+    private final AtomicBoolean shutdown = new AtomicBoolean(false);
+    private final AtomicReference<Thread> stateWatcher = new AtomicReference<>();
 
     /**
      * Resolves flag values using
@@ -69,7 +73,7 @@ public void init() throws Exception {
         flagStore.init();
         final Thread stateWatcher = new Thread(() -> {
             try {
-                while (true) {
+                while (!shutdown.get()) {
                     final StorageStateChange storageStateChange =
                             flagStore.getStateQueue().take();
                     switch (storageStateChange.getStorageState()) {
@@ -105,12 +109,12 @@ public void init() throws Exception {
                     }
                 }
             } catch (InterruptedException e) {
-                log.warn("Storage state watcher interrupted", e);
-                Thread.currentThread().interrupt();
+                log.debug("Storage state watcher interrupted, most likely shutdown was invoked", e);
             }
-        });
+        }, "InProcessResolver.stateWatcher");
         stateWatcher.setDaemon(true);
         stateWatcher.start();
+        this.stateWatcher.set(stateWatcher);
     }
 
     /**
@@ -132,7 +136,17 @@ public void onError() {
      * @throws InterruptedException if stream can't be closed within deadline.
      */
     public void shutdown() throws InterruptedException {
+        if (!shutdown.compareAndSet(false, true)) {
+            log.debug("Shutdown already in progress or completed");
+            return;
+        }
         flagStore.shutdown();
+        stateWatcher.getAndUpdate(existing -> {
+            if (existing != null) {
+                existing.interrupt();
+            }
+            return null;
+        });
     }
 
     /**

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

@@ -52,6 +52,7 @@
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
+import org.awaitility.Awaitility;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
@@ -543,6 +544,35 @@ void flagSetMetadataIsOverwrittenByFlagMetadataToEvaluation() throws Exception {
         assertThat(providerEvaluation.getFlagMetadata().getString("key")).isEqualTo("expected");
     }
 
+    @Test
+    void testStateWatcherThreadIsCleanedUpDuringShutdown() throws Exception {
+        // given
+        final Map<String, FeatureFlag> flagMap = new HashMap<>();
+        flagMap.put("booleanFlag", BOOLEAN_FLAG);
+
+        var initialThreadCount = currentDaemonThreadCount();
+
+        var queue = new LinkedBlockingQueue<StorageStateChange>();
+        InProcessResolver inProcessResolver =
+                getInProcessResolverWith(new MockStorage(flagMap, queue), (event, details, metadata) -> {});
+
+        // when
+        inProcessResolver.init();
+        var threadCountAfterInit = currentDaemonThreadCount();
+        inProcessResolver.shutdown();
+
+        // then
+        assertThat(threadCountAfterInit).isGreaterThan(initialThreadCount);
+        Awaitility.await()
+                .untilAsserted(() -> assertThat(currentDaemonThreadCount()).isEqualTo(initialThreadCount));
+    }
+
+    private long currentDaemonThreadCount() {
+        return Thread.getAllStackTraces().keySet().stream()
+                .filter(Thread::isDaemon)
+                .count();
+    }
+
     private InProcessResolver getInProcessResolverWith(final FlagdOptions options, final MockStorage storage)
             throws NoSuchFieldException, IllegalAccessException {