More graceful shutdown.

Author: kinyoklion

lib/sdk/server/contract-tests/service/src/main/java/sdktest/SdkClientEntity.java

@@ -579,7 +579,7 @@ private LDConfig buildSdkConfig(SdkConfigParams params, String tag) {
 
         // Create and configure FDv1 fallback
         ComponentConfigurer<DataSource> fdv1Fallback =
-            createFDv1FallbackSynchronizer(fallbackSynchronizer, endpoints);
+            createFDv1FallbackSynchronizer(fallbackSynchronizer);
         dataSystemBuilder.fDv1FallbackSynchronizer(fdv1Fallback);
       }
 
@@ -654,11 +654,10 @@ private static SdkConfigSynchronizerParams selectFallbackSynchronizer(
 
   /**
    * Creates the FDv1 fallback synchronizer based on the selected synchronizer config.
-   * FDv1 fallback is always polling-based.
+   * FDv1 fallback is always polling-based and uses the global service endpoints configuration.
    */
   private static ComponentConfigurer<DataSource> createFDv1FallbackSynchronizer(
-      SdkConfigSynchronizerParams synchronizer,
-      ServiceEndpointsBuilder endpoints) {
+      SdkConfigSynchronizerParams synchronizer) {
 
     // FDv1 fallback is always polling-based
     PollingDataSourceBuilder fdv1Polling = Components.pollingDataSource();
@@ -669,7 +668,8 @@ private static ComponentConfigurer<DataSource> createFDv1FallbackSynchronizer(
         fdv1Polling.pollInterval(Duration.ofMillis(synchronizer.polling.pollIntervalMs));
       }
       // Note: FDv1 polling doesn't support per-source service endpoints override,
-      // so it will use the global service endpoints configuration
+      // so it will use the global service endpoints configuration (which is set
+      // by the caller before this method is invoked)
     }
     // If streaming synchronizer, use default polling interval
     // (no additional configuration needed)

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

@@ -33,8 +33,7 @@ class FDv2DataSource implements DataSource {
      */
     private static final long defaultRecoveryTimeout = 5 * 60;
 
-    private final List<DataSourceFactory<Initializer>> initializers;
-    private final SynchronizerStateManager synchronizerStateManager;
+    private final SourceStateManager sourceStateManager;
 
     private final List<ConditionFactory> conditionFactories;
 
@@ -47,6 +46,8 @@ class FDv2DataSource implements DataSource {
 
     private final LDLogger logger;
 
+    private volatile boolean closed = false;
+
     public interface DataSourceFactory<T> {
         T build();
     }
@@ -84,7 +85,6 @@ public FDv2DataSource(
         long fallbackTimeout,
         long recoveryTimeout
     ) {
-        this.initializers = initializers;
         List<SynchronizerFactoryWithState> synchronizerFactories = synchronizers
             .stream()
             .map(SynchronizerFactoryWithState::new)
@@ -102,7 +102,7 @@ public FDv2DataSource(
             // configuration.
         }
 
-        this.synchronizerStateManager = new SynchronizerStateManager(synchronizerFactories);
+        this.sourceStateManager = new SourceStateManager(synchronizerFactories, initializers);
         this.dataSourceUpdates = dataSourceUpdates;
         this.threadPriority = threadPriority;
         this.logger = logger;
@@ -113,30 +113,33 @@ public FDv2DataSource(
 
     private void run() {
         Thread runThread = new Thread(() -> {
-            if (initializers.isEmpty() && synchronizerStateManager.getAvailableSynchronizerCount() == 0) {
+            if (!sourceStateManager.hasAvailableSources()) {
                 // There are not any initializer or synchronizers, so we are at the best state that
                 // can be achieved.
                 dataSourceUpdates.updateStatus(DataSourceStatusProvider.State.VALID, null);
                 startFuture.complete(true);
                 return;
             }
-            if (!initializers.isEmpty()) {
+            if (sourceStateManager.hasInitializers()) {
                 runInitializers();
             }
-            boolean synchronizersAvailable = synchronizerStateManager.getAvailableSynchronizerCount() != 0;
+            boolean synchronizersAvailable = sourceStateManager.hasAvailableSynchronizers();
             if(!synchronizersAvailable) {
                 // If already completed by the initializers, then this will have no effect.
-                startFuture.complete(false);
-                if (!isInitialized()) {
+                if (!isInitialized() && !closed) {
+                    // If we were closed, then closing would have handled our terminal update.
                     dataSourceUpdates.updateStatus(
                         DataSourceStatusProvider.State.OFF,
+                        // If we were shutdown during initialization, then we don't need to include an error.
                         new DataSourceStatusProvider.ErrorInfo(
                             DataSourceStatusProvider.ErrorKind.UNKNOWN,
                             0,
                             "Initializers exhausted and there are no synchronizers",
                             new Date().toInstant())
                     );
                 }
+                // If already completed has no effect.
+                startFuture.complete(false);
                 return;
             }
 
@@ -145,7 +148,9 @@ private void run() {
 
             dataSourceUpdates.updateStatus(
                 DataSourceStatusProvider.State.OFF,
-                new DataSourceStatusProvider.ErrorInfo(
+                // If the data source was closed, then we just report we are OFF without an
+                // associated error.
+                closed? null : new DataSourceStatusProvider.ErrorInfo(
                     DataSourceStatusProvider.ErrorKind.UNKNOWN,
                     0,
                     "All data source acquisition methods have been exhausted.",
@@ -164,10 +169,11 @@ private void run() {
 
     private void runInitializers() {
         boolean anyDataReceived = false;
-        for (DataSourceFactory<Initializer> factory : initializers) {
+            DataSourceFactory<Initializer> factory = sourceStateManager.getNextInitializer();
+        while(factory != null) {
             try {
                 Initializer initializer = factory.build();
-                if (synchronizerStateManager.setActiveSource(initializer)) return;
+                if (sourceStateManager.setActiveSource(initializer)) return;
                 FDv2SourceResult result = initializer.run().get();
                 switch (result.getResultType()) {
                     case CHANGE_SET:
@@ -214,6 +220,7 @@ private void runInitializers() {
                         new Date().toInstant()));
                 logger.warn("Error running initializer: {}", e.toString());
             }
+            factory = sourceStateManager.getNextInitializer();
         }
         // We received data without a selector, and we have exhausted initializers, so we are going to
         // consider ourselves initialized.
@@ -232,8 +239,8 @@ private void runInitializers() {
      * @return a list of conditions to apply to the synchronizer
      */
     private List<Condition> getConditions() {
-        int availableSynchronizers = synchronizerStateManager.getAvailableSynchronizerCount();
-        boolean isPrimeSynchronizer = synchronizerStateManager.isPrimeSynchronizer();
+        int availableSynchronizers = sourceStateManager.getAvailableSynchronizerCount();
+        boolean isPrimeSynchronizer = sourceStateManager.isPrimeSynchronizer();
 
         if (availableSynchronizers == 1) {
             // If there is only 1 synchronizer, then we cannot fall back or recover, so we don't need any conditions.
@@ -252,14 +259,14 @@ private List<Condition> getConditions() {
     private void runSynchronizers() {
         // When runSynchronizers exists, no matter how it exits, the synchronizerStateManager will be closed.
         try {
-            SynchronizerFactoryWithState availableSynchronizer = synchronizerStateManager.getNextAvailableSynchronizer();
+            SynchronizerFactoryWithState availableSynchronizer = sourceStateManager.getNextAvailableSynchronizer();
 
             // We want to continue running synchronizers for as long as any are available.
             while (availableSynchronizer != null) {
                 Synchronizer synchronizer = availableSynchronizer.build();
 
                 // Returns true if shutdown.
-                if (synchronizerStateManager.setActiveSource(synchronizer)) return;
+                if (sourceStateManager.setActiveSource(synchronizer)) return;
 
                 try {
                     boolean running = true;
@@ -284,7 +291,7 @@ private void runSynchronizers() {
                                     case RECOVERY:
                                         // For recovery, we will start at the first available synchronizer.
                                         // So we reset the source index, and finding the source will start at the beginning.
-                                        synchronizerStateManager.resetSourceIndex();
+                                        sourceStateManager.resetSourceIndex();
                                         logger.debug("The data source is attempting to recover to a higher priority synchronizer.");
                                         break;
                                 }
@@ -341,12 +348,12 @@ private void runSynchronizers() {
                             // Only trigger fallback if we're not already running the FDv1 fallback synchronizer.
                             if (
                                 result.isFdv1Fallback() &&
-                                    synchronizerStateManager.hasFDv1Fallback() &&
+                                    sourceStateManager.hasFDv1Fallback() &&
                                     // This shouldn't happen in practice, an FDv1 source shouldn't request fallback
                                     // to FDv1. But if it does, then we will discard its request.
                                     !availableSynchronizer.isFDv1Fallback()
                             ) {
-                                synchronizerStateManager.fdv1Fallback();
+                                sourceStateManager.fdv1Fallback();
                                 running = false;
                             }
                         }
@@ -362,10 +369,12 @@ private void runSynchronizers() {
                     logger.warn("Error running synchronizer: {}, will try next synchronizer, or retry.", e.toString());
                     // Move to the next synchronizer.
                 }
-                availableSynchronizer = synchronizerStateManager.getNextAvailableSynchronizer();
+                availableSynchronizer = sourceStateManager.getNextAvailableSynchronizer();
             }
-        } finally {
-            synchronizerStateManager.close();
+        } catch(Exception e) {
+            logger.error("Unexpected error in DataSource: {}", e.toString());
+        }finally {
+            sourceStateManager.close();
         }
     }
 
@@ -388,11 +397,14 @@ public boolean isInitialized() {
 
     @Override
     public void close() {
+        closed = true;
         // If there is an active source, we will shut it down, and that will result in the loop handling that source
         // exiting.
         // If we do not have an active source, then the loop will check isShutdown when attempting to set one. When
         // it detects shutdown, it will exit the loop.
-        synchronizerStateManager.close();
+        sourceStateManager.close();
+
+        dataSourceUpdates.updateStatus(DataSourceStatusProvider.State.OFF, null);
 
         // If this is already set, then this has no impact.
         startFuture.complete(false);

…sdk/server/SynchronizerStateManager.java …arkly/sdk/server/SourceStateManager.javalib/sdk/server/src/main/java/com/launchdarkly/sdk/server/SynchronizerStateManager.java renamed to lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/SourceStateManager.java lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/SynchronizerStateManager.java renamed to lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/SourceStateManager.java

@@ -1,5 +1,7 @@
 package com.launchdarkly.sdk.server;
 
+import com.launchdarkly.sdk.server.datasources.Initializer;
+
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.List;
@@ -10,9 +12,11 @@
  * <p>
  * Package-private for internal use.
  */
-class SynchronizerStateManager implements Closeable {
+class SourceStateManager implements Closeable {
     private final List<SynchronizerFactoryWithState> synchronizers;
 
+    private final List<FDv2DataSource.DataSourceFactory<Initializer>> initializers;
+
     /**
      * Lock for active sources and shutdown state.
      */
@@ -23,10 +27,13 @@ class SynchronizerStateManager implements Closeable {
     /**
      * We start at -1, so finding the next synchronizer can non-conditionally increment the index.
      */
-    private int sourceIndex = -1;
+    private int synchronizerIndex = -1;
+
+    private int initializerIndex = -1;
 
-    public SynchronizerStateManager(List<SynchronizerFactoryWithState> synchronizers) {
+    public SourceStateManager(List<SynchronizerFactoryWithState> synchronizers, List<FDv2DataSource.DataSourceFactory<Initializer>> initializers) {
         this.synchronizers = synchronizers;
+        this.initializers = initializers;
     }
 
     /**
@@ -35,7 +42,7 @@ public SynchronizerStateManager(List<SynchronizerFactoryWithState> synchronizers
      */
     public void resetSourceIndex() {
         synchronized (activeSourceLock) {
-            sourceIndex = -1;
+            synchronizerIndex = -1;
         }
     }
 
@@ -74,18 +81,24 @@ public SynchronizerFactoryWithState getNextAvailableSynchronizer() {
         synchronized (activeSourceLock) {
             SynchronizerFactoryWithState factory = null;
 
+            if(isShutdown) {
+                safeClose(activeSource);
+                activeSource = null;
+                return factory;
+            }
+
             int visited = 0;
             while(visited < synchronizers.size()) {
                 // Look for the next synchronizer starting at the position after the current one. (avoiding just re-using the same synchronizer.)
-                sourceIndex++;
+                synchronizerIndex++;
 
                 // We aren't using module here because we want to keep the stored index within range instead
                 // of increasing indefinitely.
-                if(sourceIndex >= synchronizers.size()) {
-                    sourceIndex = 0;
+                if(synchronizerIndex >= synchronizers.size()) {
+                    synchronizerIndex = 0;
                 }
 
-                SynchronizerFactoryWithState candidate = synchronizers.get(sourceIndex);
+                SynchronizerFactoryWithState candidate = synchronizers.get(synchronizerIndex);
                 if (candidate.getState() == SynchronizerFactoryWithState.State.Available) {
                     factory = candidate;
                     break;
@@ -96,6 +109,31 @@ public SynchronizerFactoryWithState getNextAvailableSynchronizer() {
         }
     }
 
+    public boolean hasAvailableSources() {
+        return hasInitializers() || getAvailableSynchronizerCount() > 0;
+    }
+
+    public boolean hasInitializers() {
+        return !initializers.isEmpty();
+    }
+
+    public boolean hasAvailableSynchronizers() {
+        return getAvailableSynchronizerCount() > 0;
+    }
+
+    public FDv2DataSource.DataSourceFactory<Initializer> getNextInitializer() {
+        synchronized (activeSourceLock) {
+            if(isShutdown) {
+                return null;
+            }
+            initializerIndex++;
+            if (initializerIndex >= initializers.size()) {
+                return null;
+            }
+            return initializers.get(initializerIndex);
+        }
+    }
+
     /**
      * Determine if the currently active synchronizer is the prime (first available) synchronizer.
      * @return true if the current synchronizer is the prime synchronizer, false otherwise
@@ -104,7 +142,7 @@ public boolean isPrimeSynchronizer() {
         synchronized (activeSourceLock) {
             for (int index = 0; index < synchronizers.size(); index++) {
                 if (synchronizers.get(index).getState() == SynchronizerFactoryWithState.State.Available) {
-                    if (sourceIndex == index) {
+                    if (synchronizerIndex == index) {
                         // This is the first synchronizer that is available, and it also is the current one.
                         return true;
                     }