HTTP-121 Ensure SerializationSchema is used in thread-safe way

CHANGELOG.md

@@ -6,6 +6,10 @@
 
 -   Added support for OIDC Bearer tokens.
 
+### Fixed
+
+-   Ensured SerializationSchema is used in thread-safe way.
+
 ## [0.15.0] - 2024-07-30
 
 ### Added

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

@@ -15,6 +15,8 @@
 import com.getindata.connectors.http.LookupQueryCreator;
 import com.getindata.connectors.http.LookupQueryCreatorFactory;
 import com.getindata.connectors.http.internal.table.lookup.LookupRow;
+import com.getindata.connectors.http.internal.utils.SynchronizedSerializationSchema;
+import static com.getindata.connectors.http.internal.table.lookup.HttpLookupConnectorOptions.ASYNC_POLLING;
 import static com.getindata.connectors.http.internal.table.lookup.HttpLookupConnectorOptions.LOOKUP_REQUEST_FORMAT;
 
 /**
@@ -47,8 +49,14 @@ public LookupQueryCreator createLookupQueryCreator(
             queryFormatAwareConfiguration
         );
 
-        SerializationSchema<RowData> serializationSchema =
-            encoder.createRuntimeEncoder(null, lookupRow.getLookupPhysicalRowDataType());
+        final SerializationSchema<RowData> serializationSchema;
+        if (readableConfig.get(ASYNC_POLLING)) {
+            serializationSchema = new SynchronizedSerializationSchema<>(
+                encoder.createRuntimeEncoder(null, lookupRow.getLookupPhysicalRowDataType()));
+        } else {
+            serializationSchema =
+                encoder.createRuntimeEncoder(null, lookupRow.getLookupPhysicalRowDataType());
+        }
 
         return new GenericJsonQueryCreator(serializationSchema);
     }

src/main/java/com/getindata/connectors/http/internal/utils/SynchronizedSerializationSchema.java

@@ -0,0 +1,35 @@
+package com.getindata.connectors.http.internal.utils;
+
+import org.apache.flink.api.common.serialization.SerializationSchema;
+
+/**
+ * Decorator which ensures that underlying SerializationSchema is called in thread-safe way.
+ *
+ * @param <T> type
+ */
+public class SynchronizedSerializationSchema<T> implements SerializationSchema<T> {
+
+    private final SerializationSchema<T> delegate;
+
+    public SynchronizedSerializationSchema(SerializationSchema<T> delegate) {
+        this.delegate = delegate;
+    }
+
+    @Override
+    public void open(InitializationContext context) throws Exception {
+        doOpen(context);
+    }
+
+    private synchronized void doOpen(InitializationContext context) throws Exception {
+        this.delegate.open(context);
+    }
+
+    @Override
+    public byte[] serialize(T element) {
+        return syncSerialize(element);
+    }
+
+    private synchronized byte[] syncSerialize(T element) {
+        return delegate.serialize(element);
+    }
+}