Fix serialize and add test.

Author: mzitnik

flink-connector-clickhouse-1.17/src/main/java/org/apache/flink/connector/clickhouse/sink/ClickHouseAsyncSinkSerializer.java

@@ -16,6 +16,7 @@ public class ClickHouseAsyncSinkSerializer extends AsyncSinkWriterStateSerialize
     protected void serializeRequestToStream(ClickHousePayload clickHousePayload, DataOutputStream dataOutputStream) throws IOException {
         byte[] bytes = clickHousePayload.getPayload();
         if (bytes != null) {
+            dataOutputStream.writeInt(V1);
             dataOutputStream.writeInt(bytes.length);
             dataOutputStream.write(bytes);
         } else {
@@ -34,11 +35,16 @@ private ClickHousePayload deserializeV1(DataInputStream dataInputStream) throws
     }
 
     @Override
-    protected ClickHousePayload deserializeRequestFromStream(long version, DataInputStream dataInputStream) throws IOException {
-        if (version == V1) {
-            return deserializeV1(dataInputStream);
+    protected ClickHousePayload deserializeRequestFromStream(long requestSize, DataInputStream dataInputStream) throws IOException {
+        if  (requestSize > 0) {
+            int version = dataInputStream.readInt();
+            if (version == V1) {
+                return deserializeV1(dataInputStream);
+            } else {
+                throw new IOException("Unsupported version " + version);
+            }
         } else {
-            throw new IOException("Unsupported version: " + version);
+            throw new IOException("request size < 0");
         }
     }
 

flink-connector-clickhouse-1.17/src/test/java/org/apache/flink/connector/clickhouse/sink/ClickHouseSinkStateTests.java

@@ -0,0 +1,29 @@
+package org.apache.flink.connector.clickhouse.sink;
+
+import org.apache.flink.connector.clickhouse.data.ClickHousePayload;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+
+public class ClickHouseSinkStateTests {
+
+    @Test
+    void SerializerTest() throws Exception {
+        byte[] data = {'H', 'e', 'l', 'l', 'o', 'W', 'o', 'r', 'l', 'd'};
+        ClickHousePayload clickHousePayload = new ClickHousePayload(data);
+
+        ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        DataOutputStream dos = new DataOutputStream(baos);
+        ClickHouseAsyncSinkSerializer serializer = new ClickHouseAsyncSinkSerializer();
+        serializer.serializeRequestToStream(clickHousePayload, dos);
+        DataInputStream dis = new DataInputStream(new ByteArrayInputStream(baos.toByteArray()));
+
+        ClickHousePayload clickHousePayload1 = serializer.deserializeRequestFromStream(dos.size(), dis);
+        Assertions.assertEquals(clickHousePayload.getPayloadLength(), clickHousePayload1.getPayloadLength());
+        Assertions.assertArrayEquals(clickHousePayload.getPayload(), clickHousePayload1.getPayload());
+    }
+}

flink-connector-clickhouse-2.0.0/src/main/java/org/apache/flink/connector/clickhouse/sink/ClickHouseAsyncSinkSerializer.java

@@ -16,6 +16,7 @@ public class ClickHouseAsyncSinkSerializer extends AsyncSinkWriterStateSerialize
     protected void serializeRequestToStream(ClickHousePayload clickHousePayload, DataOutputStream dataOutputStream) throws IOException {
         byte[] bytes = clickHousePayload.getPayload();
         if (bytes != null) {
+            dataOutputStream.writeInt(V1);
             dataOutputStream.writeInt(bytes.length);
             dataOutputStream.write(bytes);
         } else {
@@ -34,11 +35,16 @@ private ClickHousePayload deserializeV1(DataInputStream dataInputStream) throws
     }
 
     @Override
-    protected ClickHousePayload deserializeRequestFromStream(long version, DataInputStream dataInputStream) throws IOException {
-        if (version == V1) {
-            return deserializeV1(dataInputStream);
+    protected ClickHousePayload deserializeRequestFromStream(long requestSize, DataInputStream dataInputStream) throws IOException {
+        if  (requestSize > 0) {
+            int version = dataInputStream.readInt();
+            if (version == V1) {
+                return deserializeV1(dataInputStream);
+            } else {
+                throw new IOException("Unsupported version " + version);
+            }
         } else {
-            throw new IOException("Unsupported version: " + version);
+            throw new IOException("request size < 0");
         }
     }
 

flink-connector-clickhouse-2.0.0/src/test/java/org/apache/flink/connector/clickhouse/sink/ClickHouseSinkStateTests.java

@@ -0,0 +1,29 @@
+package org.apache.flink.connector.clickhouse.sink;
+
+import org.apache.flink.connector.clickhouse.data.ClickHousePayload;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+
+public class ClickHouseSinkStateTests {
+
+    @Test
+    void SerializerTest() throws Exception {
+        byte[] data = {'H', 'e', 'l', 'l', 'o', 'W', 'o', 'r', 'l', 'd'};
+        ClickHousePayload clickHousePayload = new ClickHousePayload(data);
+
+        ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        DataOutputStream dos = new DataOutputStream(baos);
+        ClickHouseAsyncSinkSerializer serializer = new ClickHouseAsyncSinkSerializer();
+        serializer.serializeRequestToStream(clickHousePayload, dos);
+        DataInputStream dis = new DataInputStream(new ByteArrayInputStream(baos.toByteArray()));
+
+        ClickHousePayload clickHousePayload1 = serializer.deserializeRequestFromStream(dos.size(), dis);
+        Assertions.assertEquals(clickHousePayload.getPayloadLength(), clickHousePayload1.getPayloadLength());
+        Assertions.assertArrayEquals(clickHousePayload.getPayload(), clickHousePayload1.getPayload());
+    }
+}

flink-connector-clickhouse-2.0.0/src/test/java/org/apache/flink/connector/test/embedded/flink/EmbeddedFlinkClusterForTests.java

@@ -21,6 +21,10 @@ static int getFromEnvOrDefault(String key, int defaultValue) {
 
     public static void setUp() throws Exception {
         Configuration config = new Configuration();
+        setUp(config);
+    }
+
+    public static void setUp(Configuration config) throws Exception {
         config.set(RestOptions.PORT, REST_PORT); // web UI port (optional)
         config.set(TaskManagerOptions.NUM_TASK_SLOTS, NUM_TASK_SLOTS);
         flinkCluster = new MiniClusterWithClientResource(
@@ -31,6 +35,7 @@ public static void setUp() throws Exception {
                         .build());
         flinkCluster.before();
     }
+
     public static void tearDown() {
         if (flinkCluster != null) {
             flinkCluster.after();