Merge pull request #589 from jefftlin/flinkx-connectors-redis/fixSinkValues

[fix][connectors redis]redis sink values保留字段类型方便反序列化

Author: whiletrue

flinkx-connectors/flinkx-connector-redis/src/main/java/com/dtstack/flinkx/connector/redis/converter/RedisColumnConverter.java

@@ -18,13 +18,15 @@
 
 package com.dtstack.flinkx.connector.redis.converter;
 
+import com.dtstack.flinkx.conf.FieldConf;
 import com.dtstack.flinkx.connector.redis.conf.RedisConf;
 import com.dtstack.flinkx.connector.redis.enums.RedisDataMode;
 import com.dtstack.flinkx.connector.redis.enums.RedisDataType;
 import com.dtstack.flinkx.converter.AbstractRowConverter;
 import com.dtstack.flinkx.element.ColumnRowData;
 import com.dtstack.flinkx.element.column.StringColumn;
 import com.dtstack.flinkx.element.column.TimestampColumn;
+import com.dtstack.flinkx.util.JsonUtil;
 
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.types.logical.LogicalType;
@@ -34,7 +36,10 @@
 
 import java.text.SimpleDateFormat;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
+import java.util.Objects;
 
 import static com.dtstack.flinkx.connector.redis.options.RedisOptions.REDIS_CRITICAL_TIME;
 import static com.dtstack.flinkx.connector.redis.options.RedisOptions.REDIS_KEY_VALUE_SIZE;
@@ -141,7 +146,17 @@ private String[] getValues(ColumnRowData row) {
     }
 
     private String concatValues(ColumnRowData row) {
-        return StringUtils.join(getValues(row), redisConf.getValueFieldDelimiter());
+        List<FieldConf> columns = redisConf.getColumn();
+        Map<String, Object> fieldMap = new HashMap<>();
+        int index = 0;
+
+        for (FieldConf fieldConf : columns) {
+            if (Objects.nonNull(row.getField(index))) {
+                fieldMap.put(fieldConf.getName(), row.getField(index).getData());
+            }
+            index++;
+        }
+        return JsonUtil.toJson(fieldMap);
     }
 
     private String concatKey(ColumnRowData row) {