fix: Deliver the terminal OFF data source status when the client is closed

lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/LDClient.java

@@ -77,6 +77,10 @@ public final class LDClient implements LDClientInterface {
   // converted, because converting it would overflow and throw (and would do so after the data system starts).
   private static final Duration MAX_MILLIS_START_WAIT = Duration.ofMillis(Long.MAX_VALUE);
 
+  // On close, how long to let already-submitted shared-executor tasks finish (so a terminal data source status
+  // broadcast is delivered rather than discarded) before forcing termination so close() cannot hang.
+  private static final long SHUTDOWN_DRAIN_MILLIS = 5000;
+
   /**
    * Creates a new client instance that connects to LaunchDarkly with the default configuration.
    * <p>
@@ -507,7 +511,20 @@ public void close() throws IOException {
     if (this.bigSegmentStoreWrapper != null) {
       this.bigSegmentStoreWrapper.close();
     }
-    this.sharedExecutor.shutdownNow();
+    // Let already-submitted tasks (such as the terminal data source status broadcast emitted above during
+    // dataSystem.close()) run, so status listeners are not left without the final OFF, then force termination if
+    // they do not finish promptly so close() cannot hang on a stuck task.
+    this.sharedExecutor.shutdown();
+    try {
+      if (!this.sharedExecutor.awaitTermination(SHUTDOWN_DRAIN_MILLIS, TimeUnit.MILLISECONDS)) {
+        this.sharedExecutor.shutdownNow();
+      }
+    } catch (InterruptedException e) {
+      this.sharedExecutor.shutdownNow();
+      Thread.currentThread().interrupt();
+    }
+    // NOTE: the awaitTermination above blocks the caller, so any teardown added below this point may be delayed
+    // by up to SHUTDOWN_DRAIN_MILLIS (until the executor drains or the drain times out).
   }
 
   @Override

lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/LDClientListenersTest.java

@@ -13,7 +13,10 @@
 import com.launchdarkly.sdk.server.interfaces.FlagValueChangeEvent;
 import com.launchdarkly.sdk.server.subsystems.BigSegmentStore;
 import com.launchdarkly.sdk.server.subsystems.BigSegmentStoreTypes;
+import com.launchdarkly.sdk.server.subsystems.ClientContext;
 import com.launchdarkly.sdk.server.subsystems.ComponentConfigurer;
+import com.launchdarkly.sdk.server.subsystems.DataSource;
+import com.launchdarkly.sdk.server.subsystems.DataSourceUpdateSink;
 import com.launchdarkly.sdk.server.subsystems.DataStore;
 import com.launchdarkly.sdk.server.subsystems.PersistentDataStore;
 
@@ -22,7 +25,12 @@
 
 import java.time.Duration;
 import java.time.Instant;
+import java.util.List;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -192,7 +200,91 @@ public void dataSourceStatusProviderSendsStatusUpdates() throws Exception {
       assertThat(newStatus.getLastError(), equalTo(errorInfo));
     }
   }
-
+
+  @Test
+  public void statusListenerReceivesTheTerminalOffWhenCloseOccursWhileTheExecutorThreadIsBusy() throws Exception {
+    // Regression test for SDK-3432. Data-source status broadcasts run on the single shared executor thread, and
+    // LDClient.close() calls shutdownNow() on that executor. If the thread is busy when the terminal OFF is
+    // broadcast on close, the OFF task is still queued and shutdownNow() discards it, so a status listener never
+    // observes the OFF the data source emits when the client closes. This test fails on that bug and passes once
+    // close() delivers the terminal status before tearing the executor down.
+    CountDownLatch listenerBlocking = new CountDownLatch(1);
+    List<DataSourceStatusProvider.State> received = new CopyOnWriteArrayList<>();
+
+    OffOnCloseDataSourceFactory factory = new OffOnCloseDataSourceFactory();
+    LDConfig config = baseConfig()
+        .dataSource(factory)
+        .events(Components.noEvents())
+        .build();
+    LDClient client = new LDClient(SDK_KEY, config);
+
+    client.getDataSourceStatusProvider().addStatusListener(status -> {
+      received.add(status.getState());
+      if (status.getState() == DataSourceStatusProvider.State.INTERRUPTED) {
+        // Occupy the single shared executor thread briefly, so the OFF that close() triggers is queued behind
+        // this callback when close() begins tearing the executor down, then free the thread so a graceful drain
+        // can still run the queued OFF (an abrupt shutdownNow discards it instead).
+        listenerBlocking.countDown();
+        try {
+          Thread.sleep(300);
+        } catch (InterruptedException e) {
+          Thread.currentThread().interrupt();
+        }
+      }
+    });
+
+    factory.sink.updateStatus(DataSourceStatusProvider.State.INTERRUPTED, new DataSourceStatusProvider.ErrorInfo(
+        DataSourceStatusProvider.ErrorKind.NETWORK_ERROR, 0, "interrupted", Instant.now()));
+    assertTrue("listener did not start", listenerBlocking.await(2, TimeUnit.SECONDS));
+
+    // close() -> data source close() broadcasts OFF (queued behind the still-busy listener). An abrupt
+    // shutdownNow() discards that queued task; a graceful drain lets it run once the thread frees.
+    client.close();
+
+    Thread.sleep(500); // allow the terminal OFF broadcast to be delivered if close() preserves it
+    assertTrue("listener should have observed INTERRUPTED",
+        received.contains(DataSourceStatusProvider.State.INTERRUPTED));
+    assertTrue("the terminal OFF emitted on close must be delivered, not dropped by shutdownNow()",
+        received.contains(DataSourceStatusProvider.State.OFF));
+  }
+
+  // A data source that becomes valid on start and, like the real streaming/polling sources, broadcasts OFF when
+  // it is closed. Exposes its update sink so a test can drive status changes.
+  static final class OffOnCloseDataSourceFactory implements ComponentConfigurer<DataSource> {
+    volatile DataSourceUpdateSink sink;
+
+    @Override
+    public DataSource build(ClientContext clientContext) {
+      sink = clientContext.getDataSourceUpdateSink();
+      return new OffOnCloseDataSource(sink);
+    }
+  }
+
+  static final class OffOnCloseDataSource implements DataSource {
+    private final DataSourceUpdateSink sink;
+
+    OffOnCloseDataSource(DataSourceUpdateSink sink) {
+      this.sink = sink;
+    }
+
+    @Override
+    public Future<Void> start() {
+      sink.updateStatus(DataSourceStatusProvider.State.VALID, null);
+      return CompletableFuture.completedFuture(null);
+    }
+
+    @Override
+    public boolean isInitialized() {
+      return true;
+    }
+
+    @Override
+    public void close() {
+      sink.updateStatus(DataSourceStatusProvider.State.OFF, new DataSourceStatusProvider.ErrorInfo(
+          DataSourceStatusProvider.ErrorKind.ERROR_RESPONSE, 0, "closed", Instant.now()));
+    }
+  }
+
   @Test
   public void dataStoreStatusMonitoringIsDisabledForInMemoryStore() throws Exception {
     LDConfig config = baseConfig()