tolerate unknown datastream envelope fields (SCH-7065)

src/main/java/com/schematic/api/datastream/DataStreamMessages.java

@@ -1,5 +1,6 @@
 package com.schematic.api.datastream;
 
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.databind.JsonNode;
 import java.util.Map;
@@ -88,6 +89,8 @@ public static MessageType fromString(String value) {
     }
 
     /** Request message sent to the datastream server. */
+    // The server may add envelope fields (e.g. stream_id); never reject a message for one.
+    @JsonIgnoreProperties(ignoreUnknown = true)
     public static class DataStreamReq {
         @JsonProperty("action")
         private final Action action;
@@ -118,6 +121,8 @@ public Map<String, String> getKeys() {
     }
 
     /** Wrapper for request messages. */
+    // The server may add envelope fields (e.g. stream_id); never reject a message for one.
+    @JsonIgnoreProperties(ignoreUnknown = true)
     public static class DataStreamBaseReq {
         @JsonProperty("data")
         private final DataStreamReq data;
@@ -132,6 +137,8 @@ public DataStreamReq getData() {
     }
 
     /** Response message received from the datastream server. */
+    // The server may add envelope fields (e.g. stream_id); never reject a message for one.
+    @JsonIgnoreProperties(ignoreUnknown = true)
     public static class DataStreamResp {
         @JsonProperty("data")
         private JsonNode data;
@@ -145,8 +152,16 @@ public static class DataStreamResp {
         @JsonProperty("message_type")
         private String messageType;
 
+        /** Server-side stream ID of the message; absent on snapshots. Recorded for replay on reconnect. */
+        @JsonProperty("stream_id")
+        private String streamId;
+
         public DataStreamResp() {}
 
+        public String getStreamId() {
+            return streamId;
+        }
+
         public JsonNode getData() {
             return data;
         }
@@ -173,6 +188,8 @@ public String getMessageType() {
     }
 
     /** Error message from the datastream server. */
+    // The server may add envelope fields (e.g. stream_id); never reject a message for one.
+    @JsonIgnoreProperties(ignoreUnknown = true)
     public static class DataStreamError {
         @JsonProperty("error")
         private String error;

src/test/java/com/schematic/api/datastream/DataStreamMessagesTest.java

@@ -0,0 +1,35 @@
+package com.schematic.api.datastream;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.schematic.api.datastream.DataStreamMessages.DataStreamError;
+import com.schematic.api.datastream.DataStreamMessages.DataStreamResp;
+import org.junit.jupiter.api.Test;
+
+public class DataStreamMessagesTest {
+    private final ObjectMapper mapper = new ObjectMapper();
+
+    @Test
+    public void responseParsesStreamIdAndIgnoresUnknownEnvelopeFields() throws Exception {
+        String json = "{\"data\":{\"id\":\"flag_1\"},\"entity_id\":\"flag_1\",\"entity_type\":\"rulesengine.Flag\","
+                + "\"message_type\":\"full\",\"stream_id\":\"1725494400000-0\",\"some_future_field\":true}";
+
+        DataStreamResp resp = mapper.readValue(json, DataStreamResp.class);
+
+        assertEquals("flag_1", resp.getEntityId());
+        assertEquals("rulesengine.Flag", resp.getEntityType());
+        assertEquals("full", resp.getMessageType());
+        assertEquals("1725494400000-0", resp.getStreamId());
+        assertEquals("flag_1", resp.getData().get("id").asText());
+    }
+
+    @Test
+    public void errorIgnoresUnknownEnvelopeFields() throws Exception {
+        String json = "{\"error\":\"not found\",\"entity_type\":\"rulesengine.Company\",\"stream_id\":\"x\"}";
+
+        DataStreamError err = mapper.readValue(json, DataStreamError.class);
+
+        assertEquals("not found", err.getError());
+    }
+}