HTTP202 Amend deserialize to work with flink 2

Signed-off-by: davidradl <[email protected]>

Author: davidradl

CHANGELOG.md

@@ -3,6 +3,7 @@
 ## [Unreleased]
 
 - allow format options to be applied to the http response decoding.
+- change deserialize method so it can work with Flink 2
 
 ## [0.24.0] - 2025-11-26
 

src/main/java/com/getindata/connectors/http/internal/table/lookup/JavaNetHttpPollingClient.java

@@ -25,6 +25,7 @@
 import org.apache.flink.configuration.ReadableConfig;
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.util.Collector;
 import org.apache.flink.util.ConfigurationException;
 import org.apache.flink.util.StringUtils;
 
@@ -269,9 +270,19 @@ private Collection<RowData> deserialize(String responseBody) throws IOException
     }
 
     private List<RowData> deserializeSingleValue(byte[] rawBytes) throws IOException {
-        return Optional.ofNullable(responseBodyDecoder.deserialize(rawBytes))
-            .map(Collections::singletonList)
-            .orElse(Collections.emptyList());
+        List<RowData> result = new ArrayList<>();
+        responseBodyDecoder.deserialize(rawBytes, new Collector<>() {
+            @Override
+            public void collect(RowData record) {
+                result.add(record);
+            }
+
+            @Override
+            public void close() {
+                // No-op
+            }
+        });
+        return result;
     }
 
     private List<RowData> deserializeArray(byte[] rawBytes) throws IOException {
@@ -281,11 +292,10 @@ private List<RowData> deserializeArray(byte[] rawBytes) throws IOException {
         List<RowData> result = new ArrayList<>();
         for (JsonNode rawObject : rawObjects) {
             if (!(rawObject instanceof NullNode)) {
-                RowData deserialized =
-                    responseBodyDecoder.deserialize(rawObject.toString().getBytes());
-                // deserialize() returns null if deserialization fails
-                if (deserialized != null) {
-                    result.add(deserialized);
+                List<RowData> deserialized = deserializeSingleValue(rawObject.toString().getBytes());
+                // deserialize() may return empty list if deserialization fails
+                if (deserialized != null && !deserialized.isEmpty()) {
+                    result.addAll(deserialized);
                 }
             }
         }