【fix】add charsetName

Author: FlechazoW

core/src/main/java/com/dtstack/flink/sql/format/dtnest/DtNestRowDeserializationSchema.java

@@ -53,13 +53,13 @@ public class DtNestRowDeserializationSchema extends AbstractDeserializationSchem
 
     private final ObjectMapper objectMapper = new ObjectMapper();
 
-    private Map<String, String> rowAndFieldMapping;
-    private Map<String, JsonNode> nodeAndJsonNodeMapping = Maps.newHashMap();
+    private final Map<String, String> rowAndFieldMapping;
+    private final Map<String, JsonNode> nodeAndJsonNodeMapping = Maps.newHashMap();
 
     private final String[] fieldNames;
     private final TypeInformation<?>[] fieldTypes;
-    private List<AbstractTableInfo.FieldExtraInfo> fieldExtraInfos;
-    private String charsetName;
+    private final List<AbstractTableInfo.FieldExtraInfo> fieldExtraInfos;
+    private final String charsetName;
 
     public DtNestRowDeserializationSchema(TypeInformation<Row> typeInfo, Map<String, String> rowAndFieldMapping,
                                           List<AbstractTableInfo.FieldExtraInfo> fieldExtraInfos,

kafka-base/kafka-base-source/src/main/java/com/dtstack/flink/sql/source/kafka/table/KafkaSourceParser.java

@@ -53,6 +53,7 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
         kafkaSourceTableInfo.setOffsetReset(MathUtil.getString(props.getOrDefault(KafkaSourceTableInfo.OFFSETRESET_KEY.toLowerCase(), EKafkaOffset.LATEST.name().toLowerCase())));
         kafkaSourceTableInfo.setTopicIsPattern(MathUtil.getBoolean(props.get(KafkaSourceTableInfo.TOPICISPATTERN_KEY.toLowerCase())));
         kafkaSourceTableInfo.setTimeZone(MathUtil.getString(props.get(KafkaSourceTableInfo.TIME_ZONE_KEY.toLowerCase())));
+        kafkaSourceTableInfo.setCharsetName(MathUtil.getString(props.getOrDefault(KafkaSourceTableInfo.CHARSET_NAME_KEY.toLowerCase(),"UTF-8")));
 
         kafkaSourceTableInfo.setSchemaString(MathUtil.getString(props.get(KafkaSourceTableInfo.SCHEMA_STRING_KEY.toLowerCase())));
         kafkaSourceTableInfo.setFieldDelimiter(MathUtil.getString(props.getOrDefault(KafkaSourceTableInfo.CSV_FIELD_DELIMITER_KEY.toLowerCase(), "|")));

kafka-base/kafka-base-source/src/main/java/com/dtstack/flink/sql/source/kafka/table/KafkaSourceTableInfo.java

@@ -54,6 +54,8 @@ public class KafkaSourceTableInfo extends AbstractSourceTableInfo {
 
     public static final String SOURCE_DATA_TYPE_KEY = "sourceDataType";
 
+    public static final String CHARSET_NAME_KEY = "charsetName";
+
     private String bootstrapServers;
 
     private String topic;