Merge branch 'v1.8.0_dev' of ssh://gitlab.prod.dtstack.cn:10022/dt-insight-engine/flinkstreamsql into v1.8.0_dev

Author: todd5167

docs/kafkaSink.md

@@ -41,7 +41,7 @@ CREATE TABLE tableName(
 |partitionKeys | 用来分区的字段|否||
 |updateMode | 回溯流数据下发模式，append,upsert.upsert模式下会将是否为回溯信息以字段形式进行下发。|否|append|
 |sinkdatatype | 写入kafka数据格式，json,avro,csv|否|json|
-|fieldDelimiter | csv数据分隔符|否| \ |
+|fieldDelimiter | csv数据分隔符|否| , |
 
 
 **kafka相关参数可以自定义，使用kafka.开头即可。**

kafka-base/kafka-base-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/AbstractKafkaProducerFactory.java

@@ -26,12 +26,9 @@
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.common.serialization.SerializationSchema;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.formats.avro.AvroRowSerializationSchema;
-import org.apache.flink.formats.csv.CsvRowSerializationSchema;
 import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
 import org.apache.flink.table.runtime.types.CRow;
-import org.apache.flink.types.Row;
 
 import java.util.Optional;
 import java.util.Properties;

kafka-base/kafka-base-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/table/KafkaSinkParser.java

@@ -49,7 +49,7 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
         }
 
         kafkaSinkTableInfo.setSchemaString(MathUtil.getString(props.get(KafkaSinkTableInfo.SCHEMA_STRING_KEY.toLowerCase())));
-        kafkaSinkTableInfo.setFieldDelimiter(MathUtil.getString(props.getOrDefault(KafkaSinkTableInfo.CSV_FIELD_DELIMITER_KEY.toLowerCase(), "|")));
+        kafkaSinkTableInfo.setFieldDelimiter(MathUtil.getString(props.getOrDefault(KafkaSinkTableInfo.CSV_FIELD_DELIMITER_KEY.toLowerCase(), ",")));
         kafkaSinkTableInfo.setBootstrapServers(MathUtil.getString(props.get(KafkaSinkTableInfo.BOOTSTRAPSERVERS_KEY.toLowerCase())));
         kafkaSinkTableInfo.setTopic(MathUtil.getString(props.get(KafkaSinkTableInfo.TOPIC_KEY.toLowerCase())));