feat: Add support for FDv1 fallback.

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

@@ -11,6 +11,7 @@
 import com.launchdarkly.sdk.server.FlagsStateOption;
 import com.launchdarkly.sdk.server.LDClient;
 import com.launchdarkly.sdk.server.LDConfig;
+import com.launchdarkly.sdk.server.interfaces.ServiceEndpoints;
 import com.launchdarkly.sdk.server.migrations.Migration;
 import com.launchdarkly.sdk.server.migrations.MigrationBuilder;
 import com.launchdarkly.sdk.server.migrations.MigrationExecution;
@@ -25,12 +26,15 @@
 import com.launchdarkly.sdk.server.integrations.HooksConfigurationBuilder;
 import com.launchdarkly.sdk.server.integrations.ServiceEndpointsBuilder;
 import com.launchdarkly.sdk.server.integrations.StreamingDataSourceBuilder;
+import com.launchdarkly.sdk.server.integrations.PollingDataSourceBuilder;
 import com.launchdarkly.sdk.server.integrations.DataSystemBuilder;
 import com.launchdarkly.sdk.server.DataSystemComponents;
 import com.launchdarkly.sdk.server.integrations.FDv2PollingInitializerBuilder;
 import com.launchdarkly.sdk.server.integrations.FDv2PollingSynchronizerBuilder;
 import com.launchdarkly.sdk.server.integrations.FDv2StreamingSynchronizerBuilder;
 import com.launchdarkly.sdk.server.interfaces.BigSegmentStoreStatusProvider;
+import com.launchdarkly.sdk.server.subsystems.ComponentConfigurer;
+import com.launchdarkly.sdk.server.subsystems.DataSource;
 import com.launchdarkly.sdk.server.subsystems.DataSourceBuilder;
 import com.launchdarkly.sdk.server.datasources.Initializer;
 import com.launchdarkly.sdk.server.datasources.Synchronizer;
@@ -563,6 +567,22 @@ private LDConfig buildSdkConfig(SdkConfigParams params, String tag) {
         }
       }
 
+      // Configure FDv1 fallback synchronizer
+      SdkConfigSynchronizerParams fallbackSynchronizer =
+          selectFallbackSynchronizer(params.dataSystem);
+      if (fallbackSynchronizer != null) {
+        // Set global polling endpoints if the fallback synchronizer has polling with custom base URI
+        if (fallbackSynchronizer.polling != null &&
+            fallbackSynchronizer.polling.baseUri != null) {
+          endpoints.polling(fallbackSynchronizer.polling.baseUri);
+        }
+
+        // Create and configure FDv1 fallback
+        ComponentConfigurer<DataSource> fdv1Fallback =
+            createFDv1FallbackSynchronizer(fallbackSynchronizer);
+        dataSystemBuilder.fDv1FallbackSynchronizer(fdv1Fallback);
+      }
+
       builder.dataSystem(dataSystemBuilder);
     }
 
@@ -601,4 +621,59 @@ private DataSourceBuilder<Synchronizer> createSynchronizer(
     }
     return null;
   }
+
+  /**
+   * Selects the best synchronizer configuration to use for FDv1 fallback.
+   * Prefers polling synchronizers, falls back to primary synchronizer.
+   */
+  private static SdkConfigSynchronizerParams selectFallbackSynchronizer(
+      SdkConfigDataSystemParams dataSystemParams) {
+
+    // Prefer secondary polling synchronizer
+    if (dataSystemParams.synchronizers != null &&
+        dataSystemParams.synchronizers.secondary != null &&
+        dataSystemParams.synchronizers.secondary.polling != null) {
+      return dataSystemParams.synchronizers.secondary;
+    }
+
+    // Fall back to primary polling synchronizer
+    if (dataSystemParams.synchronizers != null &&
+        dataSystemParams.synchronizers.primary != null &&
+        dataSystemParams.synchronizers.primary.polling != null) {
+      return dataSystemParams.synchronizers.primary;
+    }
+
+    // Fall back to primary synchronizer (even if streaming)
+    if (dataSystemParams.synchronizers != null &&
+        dataSystemParams.synchronizers.primary != null) {
+      return dataSystemParams.synchronizers.primary;
+    }
+
+    return null;
+  }
+
+  /**
+   * Creates the FDv1 fallback synchronizer based on the selected synchronizer config.
+   * FDv1 fallback is always polling-based and uses the global service endpoints configuration.
+   */
+  private static ComponentConfigurer<DataSource> createFDv1FallbackSynchronizer(
+      SdkConfigSynchronizerParams synchronizer) {
+
+    // FDv1 fallback is always polling-based
+    PollingDataSourceBuilder fdv1Polling = Components.pollingDataSource();
+
+    // Configure polling interval if the synchronizer has polling configuration
+    if (synchronizer.polling != null) {
+      if (synchronizer.polling.pollIntervalMs != null) {
+        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 (which is set
+      // by the caller before this method is invoked)
+    }
+    // If streaming synchronizer, use default polling interval
+    // (no additional configuration needed)
+
+    return fdv1Polling;
+  }
 }

lib/sdk/server/contract-tests/test-suppressions-fdv2.txt

@@ -1,4 +0,0 @@
-streaming/validation/unrecognized data that can be safely ignored/unknown event name with JSON body
-streaming/validation/unrecognized data that can be safely ignored/unknown event name with non-JSON body
-streaming/validation/unrecognized data that can be safely ignored/patch event with unrecognized path kind
-streaming/fdv2/fallback to FDv1 handling

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

@@ -0,0 +1,179 @@
+package com.launchdarkly.sdk.server;
+
+import com.launchdarkly.sdk.internal.fdv2.sources.Selector;
+import com.launchdarkly.sdk.server.datasources.FDv2SourceResult;
+import com.launchdarkly.sdk.server.datasources.Synchronizer;
+import com.launchdarkly.sdk.server.interfaces.DataSourceStatusProvider;
+import com.launchdarkly.sdk.server.interfaces.DataStoreStatusProvider;
+import com.launchdarkly.sdk.server.subsystems.DataSource;
+import com.launchdarkly.sdk.server.subsystems.DataSourceUpdateSink;
+import com.launchdarkly.sdk.server.subsystems.DataStoreTypes;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.util.AbstractMap;
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+/**
+ * Adapter that wraps a DataSource (FDv1 protocol) and exposes it as a Synchronizer (FDv2 protocol).
+ * <p>
+ * This adapter bridges the push-based DataSource interface with the pull-based Synchronizer interface
+ * by listening to updates through a custom DataSourceUpdateSink and queueing them as FDv2SourceResult objects.
+ * <p>
+ * The adapter is constructed with a factory function that receives the listening update sink and
+ * creates the DataSource. This ensures the DataSource uses the adapter's internal sink without exposing it.
+ */
+class DataSourceSynchronizerAdapter implements Synchronizer {
+    private final DataSource dataSource;
+    private final IterableAsyncQueue<FDv2SourceResult> resultQueue = new IterableAsyncQueue<>();
+    private final CompletableFuture<FDv2SourceResult> shutdownFuture = new CompletableFuture<>();
+    private final Object startLock = new Object();
+    private boolean started = false;
+    private boolean closed = false;
+    private Future<Void> startFuture;
+
+    /**
+     * Functional interface for creating a DataSource with a given update sink.
+     */
+    @FunctionalInterface
+    public interface DataSourceFactory {
+        DataSource create(DataSourceUpdateSink updateSink);
+    }
+
+    /**
+     * Creates a new adapter that wraps a DataSource.
+     *
+     * @param dataSourceFactory factory that creates the DataSource with the provided update sink
+     */
+    public DataSourceSynchronizerAdapter(DataSourceFactory dataSourceFactory) {
+        ConvertingUpdateSink convertingSink = new ConvertingUpdateSink(resultQueue);
+        this.dataSource = dataSourceFactory.create(convertingSink);
+    }
+
+    @Override
+    public CompletableFuture<FDv2SourceResult> next() {
+        synchronized (startLock) {
+            if (!started && !closed) {
+                started = true;
+                startFuture = dataSource.start();
+
+                // Monitor the start future for errors
+                // The data source will emit updates through the listening sink
+                CompletableFuture.runAsync(() -> {
+                    try {
+                        startFuture.get();
+                    } catch (ExecutionException e) {
+                        // Initialization failed - emit an interrupted status
+                        DataSourceStatusProvider.ErrorInfo errorInfo = new DataSourceStatusProvider.ErrorInfo(
+                                DataSourceStatusProvider.ErrorKind.UNKNOWN,
+                                0,
+                                e.getCause() != null ? e.getCause().toString() : e.toString(),
+                                Instant.now()
+                        );
+                        resultQueue.put(FDv2SourceResult.interrupted(errorInfo, false));
+                    } catch (CancellationException e) {
+                        // Start future was cancelled (during close) - exit cleanly
+                    } catch (InterruptedException e) {
+                        Thread.currentThread().interrupt();
+                    }
+                });
+            }
+        }
+
+        return CompletableFuture.anyOf(shutdownFuture, resultQueue.take())
+                .thenApply(result -> (FDv2SourceResult) result);
+    }
+
+    @Override
+    public void close() throws IOException {
+        synchronized (startLock) {
+            if (closed) {
+                return;
+            }
+            closed = true;
+        }
+
+        dataSource.close();
+        shutdownFuture.complete(FDv2SourceResult.shutdown());
+        if(startFuture != null) {
+            // If the start future is done, this has no effect.
+            // If it is not, then this will unblock the code waiting on start.
+            startFuture.cancel(true);
+        }
+    }
+
+    /**
+     * A DataSourceUpdateSink that converts DataSource updates into FDv2SourceResult objects.
+     * This sink doesn't delegate to any other sink - it exists solely to convert FDv1 updates to FDv2 results.
+     */
+    private static class ConvertingUpdateSink implements DataSourceUpdateSink {
+        private final IterableAsyncQueue<FDv2SourceResult> resultQueue;
+
+        public ConvertingUpdateSink(IterableAsyncQueue<FDv2SourceResult> resultQueue) {
+            this.resultQueue = resultQueue;
+        }
+
+        @Override
+        public boolean init(DataStoreTypes.FullDataSet<DataStoreTypes.ItemDescriptor> allData) {
+            // Convert the full data set into a ChangeSet and emit it
+            DataStoreTypes.ChangeSet<DataStoreTypes.ItemDescriptor> changeSet =
+                    new DataStoreTypes.ChangeSet<>(
+                            DataStoreTypes.ChangeSetType.Full,
+                            Selector.EMPTY,
+                            allData.getData(),
+                            null);
+            resultQueue.put(FDv2SourceResult.changeSet(changeSet, false));
+            return true;
+        }
+
+        @Override
+        public boolean upsert(DataStoreTypes.DataKind kind, String key, DataStoreTypes.ItemDescriptor item) {
+            // Convert the upsert into a ChangeSet with a single item and emit it
+            DataStoreTypes.KeyedItems<DataStoreTypes.ItemDescriptor> items =
+                    new DataStoreTypes.KeyedItems<>(Collections.<Map.Entry<String, DataStoreTypes.ItemDescriptor>>singletonList(
+                            new AbstractMap.SimpleEntry<>(key, item)));
+            Iterable<Map.Entry<DataStoreTypes.DataKind, DataStoreTypes.KeyedItems<DataStoreTypes.ItemDescriptor>>> data =
+                    Collections.<Map.Entry<DataStoreTypes.DataKind, DataStoreTypes.KeyedItems<DataStoreTypes.ItemDescriptor>>>singletonList(
+                            new AbstractMap.SimpleEntry<>(kind, items));
+
+            DataStoreTypes.ChangeSet<DataStoreTypes.ItemDescriptor> changeSet =
+                    new DataStoreTypes.ChangeSet<>(
+                            DataStoreTypes.ChangeSetType.Partial,
+                            Selector.EMPTY,
+                            data,
+                            null);
+            resultQueue.put(FDv2SourceResult.changeSet(changeSet, false));
+            return true;
+        }
+
+        @Override
+        public DataStoreStatusProvider getDataStoreStatusProvider() {
+            // This adapter doesn't use a data store
+            return null;
+        }
+
+        @Override
+        public void updateStatus(DataSourceStatusProvider.State newState, DataSourceStatusProvider.ErrorInfo newError) {
+            // Convert state changes to FDv2SourceResult status events
+            switch (newState) {
+                case INTERRUPTED:
+                    resultQueue.put(FDv2SourceResult.interrupted(newError, false));
+                    break;
+                case OFF:
+                    if (newError != null) {
+                        resultQueue.put(FDv2SourceResult.terminalError(newError, false));
+                    }
+                    break;
+                case VALID:
+                case INITIALIZING:
+                    // These states don't map to FDv2SourceResult status events
+                    break;
+            }
+        }
+    }
+}