Preserve headers in more error situations.

Author: kinyoklion

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

@@ -229,7 +229,7 @@ private void handleMessage(MessageEvent event) {
             fdv2Event = parseFDv2Event(eventName, event.getDataReader());
         } catch (SerializationException e) {
             logger.error("Failed to parse FDv2 event: {}", LogValues.exceptionSummary(e));
-            interruptedWithException(e, DataSourceStatusProvider.ErrorKind.INVALID_DATA);
+            interruptedWithException(e, DataSourceStatusProvider.ErrorKind.INVALID_DATA, event);
             return;
         }
 
@@ -240,7 +240,7 @@ private void handleMessage(MessageEvent event) {
         } catch (Exception e) {
             // Protocol handler threw exception processing the event - treat as invalid data
             logger.error("FDv2 protocol handler error: {}", LogValues.exceptionSummary(e));
-            interruptedWithException(e, DataSourceStatusProvider.ErrorKind.INVALID_DATA);
+            interruptedWithException(e, DataSourceStatusProvider.ErrorKind.INVALID_DATA, event);
             return;
         }
 
@@ -316,15 +316,15 @@ private void handleMessage(MessageEvent event) {
         }
     }
 
-    private void interruptedWithException(Exception e, DataSourceStatusProvider.ErrorKind kind) {
+    private void interruptedWithException(Exception e, DataSourceStatusProvider.ErrorKind kind, MessageEvent event) {
         logger.debug(LogValues.exceptionTrace(e));
         DataSourceStatusProvider.ErrorInfo errorInfo = new DataSourceStatusProvider.ErrorInfo(
                 kind,
                 0,
                 e.toString(),
                 Instant.now()
         );
-        resultQueue.put(FDv2SourceResult.interrupted(errorInfo, getFallback(e)));
+        resultQueue.put(FDv2SourceResult.interrupted(errorInfo, getFallback(event)));
         restartStream();
     }
 

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

@@ -1001,4 +1001,80 @@ public void bothFdv1FallbackAndEnvironmentIdExtractedFromHeaders() throws Except
             synchronizer.close();
         }
     }
+
+    @Test
+    public void serializationExceptionWithoutFallbackHeader() throws Exception {
+        // Verify that fallback is false when header is not present
+        // Use malformed JSON that will cause parsing to fail
+        String badEvent = makeEvent("server-intent", "{");
+
+        try (HttpServer server = HttpServer.start(Handlers.all(
+                Handlers.SSE.start(),
+                Handlers.SSE.event(badEvent),
+                Handlers.SSE.leaveOpen()))) {
+
+            HttpProperties httpProperties = toHttpProperties(clientContext("sdk-key", baseConfig().build()).getHttp());
+            SelectorSource selectorSource = mockSelectorSource();
+
+            StreamingSynchronizerImpl synchronizer = new StreamingSynchronizerImpl(
+                    httpProperties,
+                    server.getUri(),
+                    "/stream",
+                    testLogger,
+                    selectorSource,
+                    null,
+                    Duration.ofMillis(100)
+            );
+
+            CompletableFuture<FDv2SourceResult> resultFuture = synchronizer.next();
+            FDv2SourceResult result = resultFuture.get(5, TimeUnit.SECONDS);
+
+            assertNotNull(result);
+            assertEquals(FDv2SourceResult.ResultType.STATUS, result.getResultType());
+            assertEquals(FDv2SourceResult.State.INTERRUPTED, result.getStatus().getState());
+            assertEquals(DataSourceStatusProvider.ErrorKind.INVALID_DATA, result.getStatus().getErrorInfo().getKind());
+            assertEquals(false, result.isFdv1Fallback());
+
+            synchronizer.close();
+        }
+    }
+
+    @Test
+    public void serializationExceptionPreservesFallbackHeader() throws Exception {
+        // Test that when SerializationException occurs, the fallback header from the event is preserved
+        // Use definitely malformed JSON that will cause parsing to fail
+        String badEvent = makeEvent("server-intent", "{");
+
+        try (HttpServer server = HttpServer.start(Handlers.all(
+                Handlers.header("x-ld-fd-fallback", "true"),
+                Handlers.SSE.start(),
+                Handlers.SSE.event(badEvent),
+                Handlers.SSE.leaveOpen()))) {
+
+            HttpProperties httpProperties = toHttpProperties(clientContext("sdk-key", baseConfig().build()).getHttp());
+            SelectorSource selectorSource = mockSelectorSource();
+
+            StreamingSynchronizerImpl synchronizer = new StreamingSynchronizerImpl(
+                    httpProperties,
+                    server.getUri(),
+                    "/stream",
+                    testLogger,
+                    selectorSource,
+                    null,
+                    Duration.ofMillis(100)
+            );
+
+            CompletableFuture<FDv2SourceResult> resultFuture = synchronizer.next();
+            FDv2SourceResult result = resultFuture.get(5, TimeUnit.SECONDS);
+
+            assertNotNull(result);
+            assertEquals(FDv2SourceResult.ResultType.STATUS, result.getResultType());
+            assertEquals(FDv2SourceResult.State.INTERRUPTED, result.getStatus().getState());
+            assertEquals(DataSourceStatusProvider.ErrorKind.INVALID_DATA, result.getStatus().getErrorInfo().getKind());
+            assertEquals(true, result.isFdv1Fallback());
+
+            synchronizer.close();
+        }
+    }
+
 }