add kafka charsetname

Author: todd5167

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

@@ -58,17 +58,22 @@ public class DtNestRowDeserializationSchema extends AbstractDeserializationSchem
     private final String[] fieldNames;
     private final TypeInformation<?>[] fieldTypes;
     private List<TableInfo.FieldExtraInfo> fieldExtraInfos;
+    private String charsetName;
 
-    public DtNestRowDeserializationSchema(TypeInformation<Row> typeInfo, Map<String, String> rowAndFieldMapping, List<TableInfo.FieldExtraInfo> fieldExtraInfos) {
+    public DtNestRowDeserializationSchema(TypeInformation<Row> typeInfo, Map<String, String> rowAndFieldMapping,
+                                          List<TableInfo.FieldExtraInfo> fieldExtraInfos,
+                                          String charsetName) {
         this.fieldNames = ((RowTypeInfo) typeInfo).getFieldNames();
         this.fieldTypes = ((RowTypeInfo) typeInfo).getFieldTypes();
         this.rowAndFieldMapping = rowAndFieldMapping;
         this.fieldExtraInfos = fieldExtraInfos;
+        this.charsetName = charsetName;
     }
 
     @Override
     public Row deserialize(byte[] message) throws IOException {
-        JsonNode root = objectMapper.readTree(message);
+        String decoderStr = new String(message, charsetName);
+        JsonNode root = objectMapper.readTree(decoderStr);
         this.parseTree(root, null);
         Row row = new Row(fieldNames.length);
 

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

@@ -56,8 +56,8 @@ protected DeserializationMetricWrapper createDeserializationMetricWrapper(KafkaS
     private DeserializationSchema<Row> createDeserializationSchema(KafkaSourceTableInfo kafkaSourceTableInfo, TypeInformation<Row> typeInformation) {
         DeserializationSchema<Row> deserializationSchema = null;
         if (FormatType.DT_NEST.name().equalsIgnoreCase(kafkaSourceTableInfo.getSourceDataType())) {
-            deserializationSchema = new DtNestRowDeserializationSchema(typeInformation, kafkaSourceTableInfo.getPhysicalFields(), kafkaSourceTableInfo.getFieldExtraInfoList());
-
+            deserializationSchema = new DtNestRowDeserializationSchema(typeInformation, kafkaSourceTableInfo.getPhysicalFields(),
+                    kafkaSourceTableInfo.getFieldExtraInfoList(),kafkaSourceTableInfo.getCharsetName());
         } else if (FormatType.JSON.name().equalsIgnoreCase(kafkaSourceTableInfo.getSourceDataType())) {
 
             if (StringUtils.isNotBlank(kafkaSourceTableInfo.getSchemaString())) {

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

@@ -52,6 +52,7 @@ public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, O
         kafkaSourceTableInfo.setTopic(MathUtil.getString(props.get(KafkaSourceTableInfo.TOPIC_KEY.toLowerCase())));
         kafkaSourceTableInfo.setOffsetReset(MathUtil.getString(props.get(KafkaSourceTableInfo.OFFSETRESET_KEY.toLowerCase())));
         kafkaSourceTableInfo.setTopicIsPattern(MathUtil.getBoolean(props.get(KafkaSourceTableInfo.TOPICISPATTERN_KEY.toLowerCase()), false));
+        kafkaSourceTableInfo.setCharsetName(MathUtil.getString(props.getOrDefault(KafkaSourceTableInfo.CHARSET_NAME_KEY.toLowerCase(),"UTF-8")));
         kafkaSourceTableInfo.setTimeZone(MathUtil.getString(props.get(KafkaSourceTableInfo.TIME_ZONE_KEY.toLowerCase())));
         for (String key : props.keySet()) {
             if (!key.isEmpty() && key.startsWith("kafka.")) {

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

@@ -49,6 +49,8 @@ public class KafkaSourceTableInfo extends SourceTableInfo {
 
 	public static final String TOPICISPATTERN_KEY = "topicIsPattern";
 
+	public static final String CHARSET_NAME_KEY = "charsetName";
+
 	private String bootstrapServers;
 
 	private String topic;
@@ -68,6 +70,8 @@ public class KafkaSourceTableInfo extends SourceTableInfo {
 
 	private String fieldDelimiter;
 
+	private String charsetName;
+
 	public String getBootstrapServers() {
 		return bootstrapServers;
 	}
@@ -160,6 +164,14 @@ public void setFieldDelimiter(String fieldDelimiter) {
 		this.fieldDelimiter = fieldDelimiter;
 	}
 
+	public String getCharsetName() {
+		return charsetName;
+	}
+
+	public void setCharsetName(String charsetName) {
+		this.charsetName = charsetName;
+	}
+
 	@Override
 	public boolean check() {
 		Preconditions.checkNotNull(getType(), "kafka of type is required");