fix: pulsar with flink debezium-json format bug (#379)

streamnative · Jul 30, 2021 · a38b809 · a38b809
1 parent c7d1cf5
commit a38b809
Showing 1 changed file with 6 additions and 0 deletions.
diff --git a/...n/java/org/apache/flink/streaming/util/serialization/ThreadSafeDeserializationSchema.java b/...n/java/org/apache/flink/streaming/util/serialization/ThreadSafeDeserializationSchema.java
@@ -16,6 +16,7 @@
 
 import org.apache.flink.api.common.serialization.DeserializationSchema;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.util.Collector;
 
 import java.io.IOException;
 
@@ -47,6 +48,11 @@ public synchronized T deserialize(byte[] bytes) throws IOException {
         return deserializationSchema.deserialize(bytes);
     }
 
+    @Override
+    public synchronized void deserialize(byte[] message, Collector<T> out) throws IOException {
+        deserializationSchema.deserialize(message, out);
+    }
+
     @Override
     public synchronized boolean isEndOfStream(T object) {
         return deserializationSchema.isEndOfStream(object);