chore: adds DataSystem contract test support

lib/sdk/server/Makefile

@@ -9,6 +9,9 @@ test:
 	./gradlew test
 
 TEMP_TEST_OUTPUT=/tmp/sdk-test-service.log
+TEST_SERVICE_PORT ?= 8000
+SUPPRESSION_FILE=contract-tests/test-suppressions.txt
+SUPPRESSION_FILE_FDV2=contract-tests/test-suppressions-fdv2.txt
 
 # Add any extra sdk-test-harness parameters here, such as -skip for tests that are
 # temporarily not working.
@@ -18,15 +21,19 @@ build-contract-tests:
 	@cd contract-tests && ../gradlew installDist
 
 start-contract-test-service:
-	@contract-tests/service/build/install/service/bin/service
+	@PORT=$(TEST_SERVICE_PORT) contract-tests/service/build/install/service/bin/service
 
 start-contract-test-service-bg:
 	@echo "Test service output will be captured in $(TEMP_TEST_OUTPUT)"
 	@make start-contract-test-service >$(TEMP_TEST_OUTPUT) 2>&1 &
 
 run-contract-tests:
+	@echo "Running SDK contract test v2..."
 	@curl -s https://raw.githubusercontent.com/launchdarkly/sdk-test-harness/v2/downloader/run.sh \
-      | VERSION=v2 PARAMS="-url http://localhost:8000 -debug -stop-service-at-end $(TEST_HARNESS_PARAMS)" sh
+      | VERSION=v2 PARAMS="-url http://localhost:$(TEST_SERVICE_PORT) -debug -skip-from=$(SUPPRESSION_FILE) $(TEST_HARNESS_PARAMS)" sh
+	@echo "Running SDK contract test v3.0.0-alpha.1..."
+	@curl -s https://raw.githubusercontent.com/launchdarkly/sdk-test-harness/v3.0.0-alpha.1/downloader/run.sh \
+      | VERSION=v3.0.0-alpha.1 PARAMS="-url http://localhost:$(TEST_SERVICE_PORT) -debug -stop-service-at-end -skip-from=$(SUPPRESSION_FILE_FDV2) $(TEST_HARNESS_PARAMS)" sh
 
 contract-tests: build-contract-tests start-contract-test-service-bg run-contract-tests
 

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

@@ -32,6 +32,7 @@ public static class SdkConfigParams {
     SdkConfigTagParams tags;
     SdkConfigServiceEndpointParams serviceEndpoints;
     SdkConfigHookParams hooks;
+    SdkConfigDataSystemParams dataSystem;
   }
 
   public static class SdkConfigStreamParams {
@@ -73,6 +74,113 @@ public static class SdkConfigHookParams {
     List<HookConfig> hooks;
   }
 
+  /**
+   * Constants for store mode values.
+   */
+  public static class StoreMode {
+    /**
+     * Read-only mode - the data system will only read from the persistent store.
+     */
+    public static final int READ = 0;
+
+    /**
+     * Read-write mode - the data system can read from, and write to, the persistent store.
+     */
+    public static final int READ_WRITE = 1;
+  }
+
+  /**
+   * Constants for persistent store type values.
+   */
+  public static class PersistentStoreType {
+    /**
+     * Redis persistent store type.
+     */
+    public static final String REDIS = "redis";
+
+    /**
+     * DynamoDB persistent store type.
+     */
+    public static final String DYNAMODB = "dynamodb";
+
+    /**
+     * Consul persistent store type.
+     */
+    public static final String CONSUL = "consul";
+  }
+
+  /**
+   * Constants for persistent cache mode values.
+   */
+  public static class PersistentCacheMode {
+    /**
+     * Cache disabled mode.
+     */
+    public static final String OFF = "off";
+
+    /**
+     * Time-to-live cache mode with a specified TTL.
+     */
+    public static final String TTL = "ttl";
+
+    /**
+     * Infinite cache mode - cache forever.
+     */
+    public static final String INFINITE = "infinite";
+  }
+
+  public static class SdkConfigDataSystemParams {
+    SdkConfigDataStoreParams store;
+    Integer storeMode;
+    SdkConfigDataInitializerParams[] initializers;
+    SdkConfigSynchronizersParams synchronizers;
+    String payloadFilter;
+  }
+
+  public static class SdkConfigDataStoreParams {
+    SdkConfigPersistentDataStoreParams persistentDataStore;
+  }
+
+  public static class SdkConfigPersistentDataStoreParams {
+    SdkConfigPersistentStoreParams store;
+    SdkConfigPersistentCacheParams cache;
+  }
+
+  public static class SdkConfigPersistentStoreParams {
+    String type;
+    String prefix;
+    String dsn;
+  }
+
+  public static class SdkConfigPersistentCacheParams {
+    String mode;
+    Integer ttl;
+  }
+
+  public static class SdkConfigDataInitializerParams {
+    SdkConfigPollingParams polling;
+  }
+
+  public static class SdkConfigSynchronizersParams {
+    SdkConfigSynchronizerParams primary;
+    SdkConfigSynchronizerParams secondary;
+  }
+
+  public static class SdkConfigSynchronizerParams {
+    SdkConfigStreamingParams streaming;
+    SdkConfigPollingParams polling;
+  }
+
+  public static class SdkConfigPollingParams {
+    URI baseUri;
+    Long pollIntervalMs;
+  }
+
+  public static class SdkConfigStreamingParams {
+    URI baseUri;
+    Long initialRetryDelayMs;
+  }
+
   public static class HookConfig {
     String name;
     URI callbackUri;

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

@@ -25,7 +25,16 @@
 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.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.DataSourceBuilder;
+import com.launchdarkly.sdk.server.datasources.Initializer;
+import com.launchdarkly.sdk.server.datasources.Synchronizer;
+import com.launchdarkly.sdk.server.subsystems.DataSystemConfiguration;
 
 import org.jetbrains.annotations.NotNull;
 import org.slf4j.Logger;
@@ -55,6 +64,12 @@
 import sdktest.Representations.HookConfig;
 import sdktest.Representations.SdkConfigHookParams;
 import sdktest.Representations.SdkConfigParams;
+import sdktest.Representations.SdkConfigDataSystemParams;
+import sdktest.Representations.SdkConfigDataInitializerParams;
+import sdktest.Representations.SdkConfigSynchronizersParams;
+import sdktest.Representations.SdkConfigSynchronizerParams;
+import sdktest.Representations.SdkConfigPollingParams;
+import sdktest.Representations.SdkConfigStreamingParams;
 import sdktest.Representations.SecureModeHashParams;
 import sdktest.Representations.SecureModeHashResponse;
 
@@ -465,6 +480,125 @@ private LDConfig buildSdkConfig(SdkConfigParams params, String tag) {
       builder.hooks(Components.hooks().setHooks(hookList));
     }
 
+    if (params.dataSystem != null) {
+      DataSystemBuilder dataSystemBuilder = Components.dataSystem().custom();
+
+      // TODO: enable this code in the future and determine which dependencies on persistent stores need to be added to contract test build process
+      // Configure persistent store if provided
+      // if (params.dataSystem.store != null && params.dataSystem.store.persistentDataStore != null) {
+      //   var storeConfig = params.dataSystem.store.persistentDataStore;
+      //   var storeType = storeConfig.store.type.toLowerCase();
+      //   ComponentConfigurer<DataStore> persistentStore = null;
+      //
+      //   switch (storeType) {
+      //     case "redis":
+      //       // Redis store configuration
+      //       break;
+      //     case "dynamodb":
+      //       // DynamoDB store configuration
+      //       break;
+      //     case "consul":
+      //       // Consul store configuration
+      //       break;
+      //   }
+      //
+      //   if (persistentStore != null) {
+      //     // Configure cache
+      //     var cacheMode = storeConfig.cache != null ? storeConfig.cache.mode.toLowerCase() : null;
+      //     // ... cache configuration ...
+      //
+      //     // Determine store mode
+      //     var storeMode = params.dataSystem.storeMode == 0
+      //         ? DataSystemConfiguration.DataStoreMode.READ_ONLY
+      //         : DataSystemConfiguration.DataStoreMode.READ_WRITE;
+      //
+      //     dataSystemBuilder.persistentStore(persistentStore, storeMode);
+      //   }
+      // }
+
+      // Configure initializers
+      if (params.dataSystem.initializers != null && params.dataSystem.initializers.length > 0) {
+        List<DataSourceBuilder<Initializer>> initializers = new ArrayList<>();
+        for (SdkConfigDataInitializerParams initializer : params.dataSystem.initializers) {
+          if (initializer.polling != null) {
+            FDv2PollingInitializerBuilder pollingBuilder = DataSystemComponents.pollingInitializer();
+            if (initializer.polling.baseUri != null) {
+              ServiceEndpointsBuilder endpointOverride = Components.serviceEndpoints().polling(initializer.polling.baseUri);
+              pollingBuilder.serviceEndpointsOverride(endpointOverride);
+            }
+            // Note: pollInterval is not available for initializers, only for synchronizers
+            if (params.dataSystem.payloadFilter != null && !params.dataSystem.payloadFilter.isEmpty()) {
+              pollingBuilder.payloadFilter(params.dataSystem.payloadFilter);
+            }
+            initializers.add(pollingBuilder);
+          }
+        }
+        if (!initializers.isEmpty()) {
+          dataSystemBuilder.initializers(initializers.toArray(new DataSourceBuilder[0]));
+        }
+      }
+
+      // Configure synchronizers
+      if (params.dataSystem.synchronizers != null) {
+        List<DataSourceBuilder<Synchronizer>> synchronizers = new ArrayList<>();
+
+        // Primary synchronizer
+        if (params.dataSystem.synchronizers.primary != null) {
+          DataSourceBuilder<Synchronizer> primary = createSynchronizer(params.dataSystem.synchronizers.primary, params.dataSystem.payloadFilter);
+          if (primary != null) {
+            synchronizers.add(primary);
+          }
+        }
+
+        // Secondary synchronizer (optional)
+        if (params.dataSystem.synchronizers.secondary != null) {
+          DataSourceBuilder<Synchronizer> secondary = createSynchronizer(params.dataSystem.synchronizers.secondary, params.dataSystem.payloadFilter);
+          if (secondary != null) {
+            synchronizers.add(secondary);
+          }
+        }
+
+        if (!synchronizers.isEmpty()) {
+          dataSystemBuilder.synchronizers(synchronizers.toArray(new DataSourceBuilder[0]));
+        }
+      }
+
+      builder.dataSystem(dataSystemBuilder);
+    }
+
     return builder.build();
   }
+
+  private DataSourceBuilder<Synchronizer> createSynchronizer(
+      SdkConfigSynchronizerParams synchronizer,
+      String payloadFilter) {
+    if (synchronizer.polling != null) {
+      FDv2PollingSynchronizerBuilder pollingBuilder = DataSystemComponents.pollingSynchronizer();
+      if (synchronizer.polling.baseUri != null) {
+        ServiceEndpointsBuilder endpointOverride = Components.serviceEndpoints().polling(synchronizer.polling.baseUri);
+        pollingBuilder.serviceEndpointsOverride(endpointOverride);
+      }
+      if (synchronizer.polling.pollIntervalMs != null) {
+        pollingBuilder.pollInterval(Duration.ofMillis(synchronizer.polling.pollIntervalMs));
+      }
+      if (payloadFilter != null && !payloadFilter.isEmpty()) {
+        pollingBuilder.payloadFilter(payloadFilter);
+      }
+      return pollingBuilder;
+    } else if (synchronizer.streaming != null) {
+      FDv2StreamingSynchronizerBuilder streamingBuilder = DataSystemComponents.streamingSynchronizer();
+      if (synchronizer.streaming.baseUri != null) {
+        ServiceEndpointsBuilder endpointOverride = Components.serviceEndpoints().streaming(synchronizer.streaming.baseUri);
+        streamingBuilder.serviceEndpointsOverride(endpointOverride);
+      }
+      if (synchronizer.streaming.initialRetryDelayMs != null) {
+        streamingBuilder.initialReconnectDelay(Duration.ofMillis(synchronizer.streaming.initialRetryDelayMs));
+      }
+      if (payloadFilter != null && !payloadFilter.isEmpty()) {
+        streamingBuilder.payloadFilter(payloadFilter);
+      }
+      return streamingBuilder;
+    }
+    return null;
+  }
 }

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

@@ -0,0 +1,4 @@
+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/DataSystemComponents.java

@@ -38,6 +38,7 @@ public Initializer build(DataSourceBuildInputs context) {
               toHttpProperties(context.getHttp()),
               configuredBaseUri,
               StandardEndpoints.FDV2_POLLING_REQUEST_PATH,
+              payloadFilter,
               context.getBaseLogger());
 
       return new PollingInitializerImpl(
@@ -64,6 +65,7 @@ public Synchronizer build(DataSourceBuildInputs context) {
               toHttpProperties(context.getHttp()),
               configuredBaseUri,
               StandardEndpoints.FDV2_POLLING_REQUEST_PATH,
+              payloadFilter,
               context.getBaseLogger());
 
       return new PollingSynchronizerImpl(
@@ -94,7 +96,7 @@ public Synchronizer build(DataSourceBuildInputs context) {
               StandardEndpoints.FDV2_STREAMING_REQUEST_PATH,
               context.getBaseLogger(),
               context.getSelectorSource(),
-              null,
+              payloadFilter,
               initialReconnectDelay
       );
     }

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

@@ -32,25 +32,29 @@
  */
 public class DefaultFDv2Requestor implements FDv2Requestor, Closeable {
     private static final String BASIS_QUERY_PARAM = "basis";
+    private static final String FILTER_QUERY_PARAM = "filter";
 
     private final OkHttpClient httpClient;
     private final URI pollingUri;
     private final Headers headers;
     private final LDLogger logger;
     private final Map<URI, String> etags;
+    private final String payloadFilter;
 
     /**
      * Creates a DefaultFDv2Requestor.
      *
      * @param httpProperties HTTP configuration properties
      * @param baseUri base URI for the FDv2 polling endpoint
      * @param requestPath the request path to append to the base URI (e.g., "/sdk/poll")
+     * @param payloadFilter optional payload filter to add as a query parameter 
      * @param logger logger for diagnostic output
      */
-    public DefaultFDv2Requestor(HttpProperties httpProperties, URI baseUri, String requestPath, LDLogger logger) {
+    public DefaultFDv2Requestor(HttpProperties httpProperties, URI baseUri, String requestPath, String payloadFilter, LDLogger logger) {
         this.logger = logger;
         this.pollingUri = HttpHelpers.concatenateUriPath(baseUri, requestPath);
         this.etags = new HashMap<>();
+        this.payloadFilter = payloadFilter;
 
         OkHttpClient.Builder httpBuilder = httpProperties.toHttpClientBuilder();
         this.headers = httpProperties.toHeadersBuilder().build();
@@ -69,6 +73,11 @@ public CompletableFuture<FDv2PayloadResponse> Poll(Selector selector) {
                 requestUri = HttpHelpers.addQueryParam(requestUri, BASIS_QUERY_PARAM, selector.getState());
             }
 
+            // Add payload filter query parameter if present
+            if (payloadFilter != null && !payloadFilter.isEmpty()) {
+                requestUri = HttpHelpers.addQueryParam(requestUri, FILTER_QUERY_PARAM, payloadFilter);
+            }
+
             logger.debug("Making FDv2 polling request to: {}", requestUri);
 
             // Build the HTTP request