[hotfix-34625][redis5][sink]add expire attribute for redis sink's key

Author: HiLany

docs/plugin/redisSink.md

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

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

@@ -75,6 +75,8 @@ public class RedisOutputFormat extends AbstractDtRichOutputFormat<Tuple2> {
 
     protected int timeout;
 
+    protected long keyExpiredTime;
+
     private JedisPool pool;
 
     private JedisCommands jedis;
@@ -188,7 +190,7 @@ public void writeRecord(Tuple2 record) throws IOException {
             if (field != null) {
                 value = field.toString();
             }
-            jedis.set(key.toString(), value);
+            saveKey(key.toString(), value);
         }
 
         if (outRecords.getCount() % ROW_PRINT_FREQUENCY == 0){
@@ -197,6 +199,24 @@ public void writeRecord(Tuple2 record) throws IOException {
         outRecords.inc();
     }
 
+    /**
+     * 1. save key and value.
+     * 2. set expired time for key when keyExpiredTime has been set.
+     * @param key
+     * @param value
+     */
+    private void saveKey(String key, String value) {
+        if (keyExpiredTime != 0L) {
+            boolean keyExist = jedis.exists(key);
+            if (keyExist) {
+                jedis.del(key);
+            }
+            jedis.set(key, value, "NX", "PX", keyExpiredTime);
+        } else {
+            jedis.set(key, value);
+        }
+    }
+
     @Override
     public void close() throws IOException {
         if (jedisSentinelPool != null) {
@@ -289,6 +309,11 @@ public RedisOutputFormatBuilder setMasterName(String masterName){
             return this;
         }
 
+        public RedisOutputFormatBuilder setKeyExpiredTime(long 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 long 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;
     }
 
@@ -120,7 +123,8 @@ public DataStreamSink<Tuple2<Boolean, Row>> consumeDataStream(DataStream<Tuple2<
                 .setMaxTotal(this.maxTotal)
                 .setMaxIdle(this.maxIdle)
                 .setMinIdle(this.minIdle)
-                .setMasterName(this.masterName);
+                .setMasterName(this.masterName)
+                .setKeyExpiredTime(this.keyExpiredTime);
         RedisOutputFormat redisOutputFormat = builder.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(Long.parseLong(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 long keyExpiredTime;
+
     public String getUrl() {
         return url;
     }
@@ -153,6 +157,14 @@ public void setMasterName(String masterName) {
         this.masterName = masterName;
     }
 
+    public long getKeyExpiredTime() {
+        return keyExpiredTime;
+    }
+
+    public void setKeyExpiredTime(long keyExpiredTime) {
+        this.keyExpiredTime = keyExpiredTime;
+    }
+
     @Override
     public boolean check() {
         Preconditions.checkNotNull(url, "redis field of URL is required");