Fix the processing bug of abnormal parsing method of kafkaSource form…

…at. (apache#4687)
ruanwenjun · Apr 28, 2023 · 228257b · 228257b
1 parent b491020
commit 228257b
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 9 deletions.
diff --git a/...unnel-api/src/main/java/org/apache/seatunnel/api/serialization/DeserializationSchema.java b/...unnel-api/src/main/java/org/apache/seatunnel/api/serialization/DeserializationSchema.java
@@ -35,13 +35,9 @@ public interface DeserializationSchema<T> extends Serializable {
     T deserialize(byte[] message) throws IOException;
 
     default void deserialize(byte[] message, Collector<T> out) throws IOException {
-        try {
-            T deserialize = deserialize(message);
-            if (deserialize != null) {
-                out.collect(deserialize);
-            }
-        } catch (IOException e) {
-            throw new IOException(e);
+        T deserialize = deserialize(message);
+        if (deserialize != null) {
+            out.collect(deserialize);
         }
     }
 

diff --git a/...c/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceReader.java b/...c/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceReader.java
@@ -152,13 +152,13 @@ public void pollNext(Collector<SeaTunnelRow> output) throws Exception {
                                                         try {
                                                             deserializationSchema.deserialize(
                                                                     record.value(), output);
-                                                        } catch (IOException e) {
+                                                        } catch (Exception e) {
                                                             if (this.messageFormatErrorHandleWay
                                                                     == MessageFormatErrorHandleWay
                                                                             .SKIP) {
                                                                 log.warn(
                                                                         "Deserialize message failed, skip this message, message: {}",
-                                                                        record.value());
+                                                                        new String(record.value()));
                                                                 continue;
                                                             }
                                                             throw e;