Merge branch 'hotfix_1.10_4.0.x_34265' into 1.10_release_4.0.x

Author: gituser

docs/plugin/redisSink.md

@@ -44,6 +44,7 @@ redis5.0
 |minIdle|最小空闲连接数|否||0|
 |masterName| 哨兵模式下的masterName|否||
 |primarykeys|主键字段，多个字段以逗号分割|是||
+|keyExpiredTime|redis sink的key的过期时间。默认是0（永不过期），单位是s。|否||
       
 
 ## 5.样例：

redis5/redis5-sink/src/main/java/com/dtstack/flink/sql/sink/redis/RedisOutputFormat.java

@@ -47,19 +47,35 @@
  */
 public class RedisOutputFormat extends AbstractDtRichOutputFormat<Tuple2> {
     private static final Logger LOG = LoggerFactory.getLogger(RedisOutputFormat.class);
+
     protected String[] fieldNames;
+
     protected TypeInformation<?>[] fieldTypes;
+
     protected List<String> primaryKeys;
+
     protected int timeout = 10000;
+
     private String url;
+
     private String database = "0";
+
     private String tableName;
+
     private String password;
+
     private int redisType;
+
     private String maxTotal;
+
     private String maxIdle;
+
     private String minIdle;
+
     private String masterName;
+
+    protected int keyExpiredTime;
+
     private JedisPool pool;
 
     private JedisCommands jedis;
@@ -146,15 +162,31 @@ public void writeRecord(Tuple2 record) throws IOException {
         for (int i = 0; i < fieldNames.length; i++) {
             refData.put(fieldNames[i], row.getField(i));
         }
-        String redisKey = buildCacheKey(refData);
-        refData.forEach((key, value) -> jedis.hset(redisKey, key, String.valueOf(value)));
-
+        save(refData);
         if (outRecords.getCount() % ROW_PRINT_FREQUENCY == 0) {
             LOG.info(record.toString());
         }
         outRecords.inc();
     }
 
+
+    /**
+     *  1. build key from map.
+     *  2. save key and value.
+     *  3. set expired time for key when keyExpiredTime has been set.
+     * @param refData
+     */
+    private synchronized void save(Map<String, Object> refData) {
+        String key = buildCacheKey(refData);
+        try {
+            refData.forEach((field, value) -> jedis.hset(key, field, String.valueOf(value)));
+        } finally {
+            if (keyExpiredTime != 0) {
+                jedis.expire(key, keyExpiredTime);
+            }
+        }
+    }
+
     @Override
     public void close() throws IOException {
         if (jedisSentinelPool != null) {
@@ -254,6 +286,11 @@ public RedisOutputFormatBuilder setMasterName(String masterName) {
             return this;
         }
 
+        public RedisOutputFormatBuilder setKeyExpiredTime(int keyExpiredTime){
+            redisOutputFormat.keyExpiredTime = keyExpiredTime;
+            return this;
+        }
+
         public RedisOutputFormat finish() {
             if (redisOutputFormat.url == null) {
                 throw new IllegalArgumentException("No URL supplied.");

redis5/redis5-sink/src/main/java/com/dtstack/flink/sql/sink/redis/RedisSink.java

@@ -71,6 +71,8 @@ public class RedisSink implements RetractStreamTableSink<Row>, IStreamSinkGener<
 
     protected String registerTableName;
 
+    protected int keyExpiredTime;
+
     public RedisSink() {
 
     }
@@ -92,6 +94,7 @@ public RedisSink genStreamSink(AbstractTargetTableInfo targetTableInfo) {
         this.parallelism = Objects.isNull(redisTableInfo.getParallelism()) ?
                 parallelism : redisTableInfo.getParallelism();
         this.registerTableName = redisTableInfo.getName();
+        this.keyExpiredTime = redisTableInfo.getKeyExpiredTime();
         return this;
     }
 
@@ -121,6 +124,7 @@ public DataStreamSink<Tuple2<Boolean, Row>> consumeDataStream(DataStream<Tuple2<
                 .setMaxIdle(this.maxIdle)
                 .setMinIdle(this.minIdle)
                 .setMasterName(this.masterName)
+                .setKeyExpiredTime(this.keyExpiredTime)
                 .finish();
         RichSinkFunction richSinkFunction = new OutputFormatSinkFunction(redisOutputFormat);
         DataStreamSink dataStreamSink = dataStream.addSink(richSinkFunction)

redis5/redis5-sink/src/main/java/com/dtstack/flink/sql/sink/redis/table/RedisSinkParser.java

@@ -49,6 +49,10 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
         redisTableInfo.setRedisType(MathUtil.getString(props.get(RedisTableInfo.REDIS_TYPE.toLowerCase())));
         redisTableInfo.setMasterName(MathUtil.getString(props.get(RedisTableInfo.MASTER_NAME.toLowerCase())));
 
+        if (props.get(RedisTableInfo.KEY_EXPIRED_TIME.toLowerCase()) != null) {
+            redisTableInfo.setKeyExpiredTime(Integer.parseInt(MathUtil.getString(props.get(RedisTableInfo.KEY_EXPIRED_TIME.toLowerCase()))));
+        }
+
         String primaryKeysStr = MathUtil.getString(props.get(RedisTableInfo.PRIMARY_KEYS_NAME));
         List<String> primaryKeysList = Lists.newArrayList();
         if (!StringUtils.isEmpty(primaryKeysStr)) {

redis5/redis5-sink/src/main/java/com/dtstack/flink/sql/sink/redis/table/RedisTableInfo.java

@@ -49,6 +49,8 @@ public class RedisTableInfo extends AbstractTargetTableInfo {
 
     public static final String PRIMARY_KEYS_NAME = "primarykeys";
 
+    public static final String KEY_EXPIRED_TIME = "keyExpiredTime";
+
     public RedisTableInfo(){
         setType(CURR_TYPE);
     }
@@ -73,6 +75,8 @@ public RedisTableInfo(){
 
     private String masterName;
 
+    private int keyExpiredTime;
+
     public String getUrl() {
         return url;
     }
@@ -153,6 +157,14 @@ public void setMasterName(String masterName) {
         this.masterName = masterName;
     }
 
+    public int getKeyExpiredTime() {
+        return keyExpiredTime;
+    }
+
+    public void setKeyExpiredTime(int keyExpiredTime) {
+        this.keyExpiredTime = keyExpiredTime;
+    }
+
     @Override
     public boolean check() {
         Preconditions.checkNotNull(url, "redis field of URL is required");