From 3dd76b85634ea0043be602015dc4ac44d8e3cc33 Mon Sep 17 00:00:00 2001
From: FlechazoW <tianzhu4680@protonmail.com>
Date: Sun, 17 Jul 2022 16:58:41 +0800
Subject: [PATCH] [hotfix-#1063][kafka] Fix sync Task of kafka throws npe.

---
 .../connector/kafka/converter/KafkaColumnConverter.java       | 4 ----
 .../main/java/com/dtstack/chunjun/source/SourceFactory.java   | 2 +-
 2 files changed, 1 insertion(+), 5 deletions(-)

diff --git a/chunjun-connectors/chunjun-connector-kafka/src/main/java/com/dtstack/chunjun/connector/kafka/converter/KafkaColumnConverter.java b/chunjun-connectors/chunjun-connector-kafka/src/main/java/com/dtstack/chunjun/connector/kafka/converter/KafkaColumnConverter.java
index 4c46e61909..3edaa3e5e7 100644
--- a/chunjun-connectors/chunjun-connector-kafka/src/main/java/com/dtstack/chunjun/connector/kafka/converter/KafkaColumnConverter.java
+++ b/chunjun-connectors/chunjun-connector-kafka/src/main/java/com/dtstack/chunjun/connector/kafka/converter/KafkaColumnConverter.java
@@ -70,8 +70,6 @@ public class KafkaColumnConverter extends AbstractRowConverter<String, Object, b
 
     /** source kafka msg decode */
     private final IDecode decode;
-    /** sink json Decoder */
-    private final JsonDecoder jsonDecoder;
     /** kafka Conf */
     private final KafkaConf kafkaConf;
     /** kafka sink out fields */
@@ -80,7 +78,6 @@ public class KafkaColumnConverter extends AbstractRowConverter<String, Object, b
     public KafkaColumnConverter(KafkaConf kafkaConf, List<String> keyTypeList) {
         this.kafkaConf = kafkaConf;
         this.outList = keyTypeList;
-        this.jsonDecoder = new JsonDecoder();
         if (DEFAULT_CODEC.defaultValue().equals(kafkaConf.getCodec())) {
             this.decode = new JsonDecoder();
         } else {
@@ -90,7 +87,6 @@ public KafkaColumnConverter(KafkaConf kafkaConf, List<String> keyTypeList) {
 
     public KafkaColumnConverter(KafkaConf kafkaConf) {
         this.commonConf = this.kafkaConf = kafkaConf;
-        this.jsonDecoder = new JsonDecoder();
         if (DEFAULT_CODEC.defaultValue().equals(kafkaConf.getCodec())) {
             this.decode = new JsonDecoder();
         } else {
diff --git a/chunjun-core/src/main/java/com/dtstack/chunjun/source/SourceFactory.java b/chunjun-core/src/main/java/com/dtstack/chunjun/source/SourceFactory.java
index 1f487a9bf6..9f3142773a 100644
--- a/chunjun-core/src/main/java/com/dtstack/chunjun/source/SourceFactory.java
+++ b/chunjun-core/src/main/java/com/dtstack/chunjun/source/SourceFactory.java
@@ -127,7 +127,7 @@ protected DataStream<RowData> createInput(
     protected DataStream<RowData> createInput(
             RichParallelSourceFunction<RowData> function, String sourceName) {
         Preconditions.checkNotNull(sourceName);
-        return env.addSource(function, sourceName, getTypeInformation());
+        return env.addSource(function, sourceName);
     }
 
     protected DataStream<RowData> createInput(InputFormat<RowData, InputSplit> inputFormat) {