feat: Add FDv2 Streaming Synchronizer

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

@@ -124,8 +124,20 @@ protected CompletableFuture<FDv2SourceResult> poll(Selector selector, boolean on
                     case NONE:
                         break;
                     case INTERNAL_ERROR: {
+                        FDv2ProtocolHandler.FDv2ActionInternalError internalErrorAction = (FDv2ProtocolHandler.FDv2ActionInternalError) res;
+                        DataSourceStatusProvider.ErrorKind kind = DataSourceStatusProvider.ErrorKind.UNKNOWN;
+                        switch (internalErrorAction.getErrorType()) {
+                            case MISSING_PAYLOAD:
+                            case JSON_ERROR:
+                                kind = DataSourceStatusProvider.ErrorKind.INVALID_DATA;
+                                break;
+                            case UNKNOWN_EVENT:
+                            case IMPLEMENTATION_ERROR:
+                            case PROTOCOL_ERROR:
+                                break;
+                        }
                         DataSourceStatusProvider.ErrorInfo info = new DataSourceStatusProvider.ErrorInfo(
-                                DataSourceStatusProvider.ErrorKind.UNKNOWN,
+                                kind,
                                 0,
                                 "Internal error occurred during polling",
                                 new Date().toInstant());

lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/datasources/Initializer.java

@@ -46,6 +46,8 @@
 public interface Initializer extends Closeable {
     /**
      * Run the initializer to completion.
+     * <p>
+     * This method is intended to be called only single time for an instance.
      * @return The result of the initializer.
      */
     CompletableFuture<FDv2SourceResult> run();

lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/datasources/Synchronizer.java

@@ -52,6 +52,8 @@ public interface Synchronizer extends Closeable {
      * <p>
      * This method is intended to be driven by a single thread, and for there to be a single outstanding call
      * at any given time.
+     * <p>
+     *  Once SHUTDOWN, TERMINAL_ERROR, or GOODBYE has been produced, then no further calls to next() should be made.
      * @return a future that will complete when the next result is available
      */
     CompletableFuture<FDv2SourceResult> next();

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

@@ -329,6 +329,98 @@ public void emptyEventsArray() throws Exception {
         assertEquals(FDv2SourceResult.ResultType.STATUS, result.getResultType());
         assertEquals(FDv2SourceResult.State.TERMINAL_ERROR, result.getStatus().getState());
 
-
+
+    }
+
+    @Test
+    public void internalErrorWithInvalidDataKind() throws Exception {
+        FDv2Requestor requestor = mockRequestor();
+        SelectorSource selectorSource = mockSelectorSource();
+
+        // Create a response with malformed payload-transferred event. `state->states`.
+        // This will trigger JSON_ERROR internal error which maps to INVALID_DATA
+        String malformedPutObjectJson = "{\n" +
+            "  \"events\": [\n" +
+            "    {\n" +
+            "      \"event\": \"server-intent\",\n" +
+            "      \"data\": {\n" +
+            "        \"payloads\": [{\n" +
+            "          \"id\": \"payload-1\",\n" +
+            "          \"target\": 100,\n" +
+            "          \"intentCode\": \"xfer-full\",\n" +
+            "          \"reason\": \"payload-missing\"\n" +
+            "        }]\n" +
+            "      }\n" +
+            "    },\n" +
+            "    {\n" +
+            "      \"event\": \"payload-transferred\",\n" +
+            "      \"data\": {\n" +
+            "        \"states\": \"(p:payload-1:100)\",\n" +
+            "        \"version\": 100\n" +
+            "      }\n" +
+            "    },\n" +
+            "    {\n" +
+            "      \"event\": \"put-object\",\n" +
+            "      \"data\": {}\n" +
+            "    }\n" +
+            "  ]\n" +
+            "}";
+
+        FDv2Requestor.FDv2PayloadResponse response = new FDv2Requestor.FDv2PayloadResponse(
+            com.launchdarkly.sdk.internal.fdv2.payloads.FDv2Event.parseEventsArray(malformedPutObjectJson),
+            okhttp3.Headers.of()
+        );
+
+        when(requestor.Poll(any(Selector.class)))
+            .thenReturn(CompletableFuture.completedFuture(response));
+
+        PollingInitializerImpl initializer = new PollingInitializerImpl(requestor, testLogger, selectorSource);
+
+        CompletableFuture<FDv2SourceResult> resultFuture = initializer.run();
+        FDv2SourceResult result = resultFuture.get(5, TimeUnit.SECONDS);
+
+        assertNotNull(result);
+        assertEquals(FDv2SourceResult.ResultType.STATUS, result.getResultType());
+        assertEquals(FDv2SourceResult.State.TERMINAL_ERROR, result.getStatus().getState());
+        assertEquals(DataSourceStatusProvider.ErrorKind.INVALID_DATA, result.getStatus().getErrorInfo().getKind());
+
+
+    }
+
+    @Test
+    public void internalErrorWithUnknownKind() throws Exception {
+        FDv2Requestor requestor = mockRequestor();
+        SelectorSource selectorSource = mockSelectorSource();
+
+        // Create a response with an unrecognized event type
+        // This will trigger UNKNOWN_EVENT internal error which maps to UNKNOWN error kind
+        String unknownEventJson = "{\n" +
+            "  \"events\": [\n" +
+            "    {\n" +
+            "      \"event\": \"unrecognized-event-type\",\n" +
+            "      \"data\": {}\n" +
+            "    }\n" +
+            "  ]\n" +
+            "}";
+
+        FDv2Requestor.FDv2PayloadResponse response = new FDv2Requestor.FDv2PayloadResponse(
+            com.launchdarkly.sdk.internal.fdv2.payloads.FDv2Event.parseEventsArray(unknownEventJson),
+            okhttp3.Headers.of()
+        );
+
+        when(requestor.Poll(any(Selector.class)))
+            .thenReturn(CompletableFuture.completedFuture(response));
+
+        PollingInitializerImpl initializer = new PollingInitializerImpl(requestor, testLogger, selectorSource);
+
+        CompletableFuture<FDv2SourceResult> resultFuture = initializer.run();
+        FDv2SourceResult result = resultFuture.get(5, TimeUnit.SECONDS);
+
+        assertNotNull(result);
+        assertEquals(FDv2SourceResult.ResultType.STATUS, result.getResultType());
+        assertEquals(FDv2SourceResult.State.TERMINAL_ERROR, result.getStatus().getState());
+        assertEquals(DataSourceStatusProvider.ErrorKind.UNKNOWN, result.getStatus().getErrorInfo().getKind());
+
+
     }
 }

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

@@ -722,4 +722,148 @@ public void nonRecoverableThenRecoverableErrorStopsPolling() throws Exception {
             executor.shutdown();
         }
     }
+
+    @Test
+    public void internalErrorWithInvalidDataKindContinuesPolling() throws Exception {
+        FDv2Requestor requestor = mockRequestor();
+        SelectorSource selectorSource = mockSelectorSource();
+        ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
+
+        AtomicInteger callCount = new AtomicInteger(0);
+        when(requestor.Poll(any(Selector.class))).thenAnswer(invocation -> {
+            int count = callCount.incrementAndGet();
+            if (count == 1) {
+                // First call returns response with malformed put-object which triggers INTERNAL_ERROR (INVALID_DATA)
+                String malformedPutObjectJson = "{\n" +
+                    "  \"events\": [\n" +
+                    "    {\n" +
+                    "      \"event\": \"server-intent\",\n" +
+                    "      \"data\": {\n" +
+                    "        \"payloads\": [{\n" +
+                    "          \"id\": \"payload-1\",\n" +
+                    "          \"target\": 100,\n" +
+                    "          \"intentCode\": \"xfer-full\",\n" +
+                    "          \"reason\": \"payload-missing\"\n" +
+                    "        }]\n" +
+                    "      }\n" +
+                    "    },\n" +
+                    "    {\n" +
+                    "      \"event\": \"payload-transferred\",\n" +
+                    "      \"data\": {\n" +
+                    "        \"state\": \"(p:payload-1:100)\",\n" +
+                    "        \"version\": 100\n" +
+                    "      }\n" +
+                    "    },\n" +
+                    "    {\n" +
+                    "      \"event\": \"put-object\",\n" +
+                    "      \"data\": {}\n" +
+                    "    }\n" +
+                    "  ]\n" +
+                    "}";
+
+                return CompletableFuture.completedFuture(new FDv2Requestor.FDv2PayloadResponse(
+                    com.launchdarkly.sdk.internal.fdv2.payloads.FDv2Event.parseEventsArray(malformedPutObjectJson),
+                    okhttp3.Headers.of()
+                ));
+            } else {
+                // Subsequent calls succeed
+                return CompletableFuture.completedFuture(makeSuccessResponse());
+            }
+        });
+
+        try {
+            PollingSynchronizerImpl synchronizer = new PollingSynchronizerImpl(
+                    requestor,
+                    testLogger,
+                    selectorSource,
+                    executor,
+                    Duration.ofMillis(50)
+            );
+
+            // Wait for multiple polls
+            Thread.sleep(250);
+
+            // First result should be interrupted with INVALID_DATA error kind
+            FDv2SourceResult result1 = synchronizer.next().get(1, TimeUnit.SECONDS);
+            assertNotNull(result1);
+            assertEquals(FDv2SourceResult.ResultType.STATUS, result1.getResultType());
+            assertEquals(FDv2SourceResult.State.INTERRUPTED, result1.getStatus().getState());
+            assertEquals(DataSourceStatusProvider.ErrorKind.INVALID_DATA, result1.getStatus().getErrorInfo().getKind());
+
+            // Second result should be success (polling continued)
+            FDv2SourceResult result2 = synchronizer.next().get(1, TimeUnit.SECONDS);
+            assertNotNull(result2);
+            assertEquals(FDv2SourceResult.ResultType.CHANGE_SET, result2.getResultType());
+
+            // Verify polling continued after internal error
+            assertTrue("Should have made at least 2 calls", callCount.get() >= 2);
+
+            synchronizer.close();
+        } finally {
+            executor.shutdown();
+        }
+    }
+
+    @Test
+    public void internalErrorWithUnknownKindContinuesPolling() throws Exception {
+        FDv2Requestor requestor = mockRequestor();
+        SelectorSource selectorSource = mockSelectorSource();
+        ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
+
+        AtomicInteger callCount = new AtomicInteger(0);
+        when(requestor.Poll(any(Selector.class))).thenAnswer(invocation -> {
+            int count = callCount.incrementAndGet();
+            if (count == 1) {
+                // First call returns response with unknown event which triggers INTERNAL_ERROR (UNKNOWN)
+                String unknownEventJson = "{\n" +
+                    "  \"events\": [\n" +
+                    "    {\n" +
+                    "      \"event\": \"unrecognized-event-type\",\n" +
+                    "      \"data\": {}\n" +
+                    "    }\n" +
+                    "  ]\n" +
+                    "}";
+
+                return CompletableFuture.completedFuture(new FDv2Requestor.FDv2PayloadResponse(
+                    com.launchdarkly.sdk.internal.fdv2.payloads.FDv2Event.parseEventsArray(unknownEventJson),
+                    okhttp3.Headers.of()
+                ));
+            } else {
+                // Subsequent calls succeed
+                return CompletableFuture.completedFuture(makeSuccessResponse());
+            }
+        });
+
+        try {
+            PollingSynchronizerImpl synchronizer = new PollingSynchronizerImpl(
+                    requestor,
+                    testLogger,
+                    selectorSource,
+                    executor,
+                    Duration.ofMillis(50)
+            );
+
+            // Wait for multiple polls
+            Thread.sleep(250);
+
+            // First result should be interrupted with UNKNOWN error kind
+            FDv2SourceResult result1 = synchronizer.next().get(1, TimeUnit.SECONDS);
+            assertNotNull(result1);
+            assertEquals(FDv2SourceResult.ResultType.STATUS, result1.getResultType());
+            assertEquals(FDv2SourceResult.State.INTERRUPTED, result1.getStatus().getState());
+            assertEquals(DataSourceStatusProvider.ErrorKind.UNKNOWN, result1.getStatus().getErrorInfo().getKind());
+
+            // Second result should be success (polling continued)
+            FDv2SourceResult result2 = synchronizer.next().get(1, TimeUnit.SECONDS);
+            assertNotNull(result2);
+            assertEquals(FDv2SourceResult.ResultType.CHANGE_SET, result2.getResultType());
+
+            // Verify polling continued after internal error
+            assertTrue("Should have made at least 2 calls", callCount.get() >= 2);
+
+            synchronizer.close();
+        } finally {
+            executor.shutdown();
+        }
+    }
 }