add support for expire key on time, the type of time is LocalTime, eg…

…: 10:00 12:12:01

src/main/java/org/apache/flink/streaming/connectors/redis/common/config/RedisOptions.java

@@ -164,4 +164,10 @@ private RedisOptions() {}
  .enumType(RedisValueDataStructure.class)
  .defaultValue(RedisValueDataStructure.column)
  .withDescription("Optional redis value data structure.");
+
+ public static final ConfigOption<String> EXPIRE_ON_TIME =
+ ConfigOptions.key("expire.on.time")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Optional redis key expire on time, eg: 10:00 12:12:01");
 }

src/main/java/org/apache/flink/streaming/connectors/redis/common/container/RedisClusterContainer.java

@@ -411,4 +411,21 @@ public RedisFuture<Map<String, String>> hgetAll(String key) {
  public RedisClusterAsyncCommands getAsyncCommands() {
  return clusterAsyncCommands;
  }
+
+ @Override
+ public RedisFuture<Long> getTTL(String key) {
+ RedisFuture<Long> result = null;
+ try {
+ result = redisFuture = clusterAsyncCommands.ttl(key);
+ } catch (Exception e) {
+ if (LOG.isErrorEnabled()) {
+ LOG.error(
+ "Cannot send Redis message with command ttl to key {} error message {}",
+ key,
+ e.getMessage());
+ }
+ throw e;
+ }
+ return result;
+ }
 }

src/main/java/org/apache/flink/streaming/connectors/redis/common/container/RedisCommandsContainer.java

@@ -191,6 +191,14 @@ public interface RedisCommandsContainer extends Serializable {
  */
  void expire(String key, int seconds);
 
+ /**
+ * get ttl of key.
+ *
+ * @param key
+ * @return
+ */
+ RedisFuture<Long> getTTL(String key);
+
  /**
  * delete key in map.
  *

src/main/java/org/apache/flink/streaming/connectors/redis/common/container/RedisContainer.java

@@ -427,4 +427,21 @@ public void srem(String setName, String value) {
  public RedisClusterAsyncCommands getAsyncCommands() {
  return asyncCommands;
  }
+
+ @Override
+ public RedisFuture<Long> getTTL(String key) {
+ RedisFuture<Long> result = null;
+ try {
+ result = redisFuture = asyncCommands.ttl(key);
+ } catch (Exception e) {
+ if (LOG.isErrorEnabled()) {
+ LOG.error(
+ "Cannot send Redis message with command ttl to key {} error message {}",
+ key,
+ e.getMessage());
+ }
+ throw e;
+ }
+ return result;
+ }
 }

src/main/java/org/apache/flink/streaming/connectors/redis/common/mapper/RedisCommandDescription.java

@@ -1,6 +1,7 @@
 package org.apache.flink.streaming.connectors.redis.common.mapper;
 
 import java.io.Serializable;
+import java.time.LocalTime;
 
 /** */
 public class RedisCommandDescription extends RedisCommandBaseDescription implements Serializable {
@@ -9,13 +10,19 @@ public class RedisCommandDescription extends RedisCommandBaseDescription impleme
 
  private Integer ttl;
 
- public RedisCommandDescription(RedisCommand redisCommand, Integer ttl) {
- super(redisCommand);
+ private LocalTime expireTime;
 
+ public RedisCommandDescription(RedisCommand redisCommand, Integer ttl, LocalTime expireTime) {
+ super(redisCommand);
+ this.expireTime = expireTime;
  this.ttl = ttl;
  }
 
  public Integer getTTL() {
  return ttl;
  }
+
+ public LocalTime getExpireTime() {
+ return expireTime;
+ }
 }

src/main/java/org/apache/flink/streaming/connectors/redis/common/mapper/row/sink/RowRedisSinkMapper.java

@@ -10,7 +10,9 @@
 import org.apache.flink.table.data.GenericRowData;
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.util.StringUtils;
 
+import java.time.LocalTime;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -22,6 +24,8 @@ public abstract class RowRedisSinkMapper
 
  private Integer ttl;
 
+ private LocalTime expireTime;
+
  private RedisCommand redisCommand;
 
  public RowRedisSinkMapper(int ttl, RedisCommand redisCommand) {
@@ -45,11 +49,15 @@ public RowRedisSinkMapper(RedisCommand redisCommand, Map<String, String> config)
  public RowRedisSinkMapper(RedisCommand redisCommand, ReadableConfig config) {
  this.redisCommand = redisCommand;
  this.ttl = config.get(RedisOptions.TTL);
+ String expireOnTime = config.get(RedisOptions.EXPIRE_ON_TIME);
+ if (!StringUtils.isNullOrWhitespaceOnly(expireOnTime)) {
+ this.expireTime = LocalTime.parse(expireOnTime);
+ }
  }
 
  @Override
  public RedisCommandDescription getCommandDescription() {
- return new RedisCommandDescription(redisCommand, ttl);
+ return new RedisCommandDescription(redisCommand, ttl, expireTime);
  }
 
  @Override

src/main/java/org/apache/flink/streaming/connectors/redis/table/RedisDynamicTableFactory.java

@@ -105,6 +105,7 @@ public Set<ConfigOption<?>> optionalOptions() {
  options.add(RedisOptions.VALUE_DATA_STRUCTURE);
  options.add(RedisOptions.REDIS_MASTER_NAME);
  options.add(RedisOptions.SENTINELS_INFO);
+ options.add(RedisOptions.EXPIRE_ON_TIME);
  return options;
  }
 

src/main/java/org/apache/flink/streaming/connectors/redis/table/RedisLookupFunction.java

@@ -56,9 +56,6 @@ public RedisLookupFunction(
  Preconditions.checkNotNull(
  flinkConfigBase, "Redis connection pool config should not be null");
  Preconditions.checkNotNull(redisMapper, "Redis Mapper can not be null");
- Preconditions.checkNotNull(
- redisMapper.getCommandDescription(),
- "Redis Mapper data type description can not be null");
 
  this.flinkConfigBase = flinkConfigBase;
  this.cacheTtl = redisLookupOptions.getCacheTtl();
@@ -77,6 +74,8 @@ public RedisLookupFunction(
  }
 
  RedisCommandBaseDescription redisCommandDescription = redisMapper.getCommandDescription();
+ Preconditions.checkNotNull(
+ redisCommandDescription, "Redis Mapper data type description can not be null");
  this.redisCommand = redisCommandDescription.getRedisCommand();
  Preconditions.checkArgument(
  redisCommand == RedisCommand.HGET || redisCommand == RedisCommand.GET,

src/main/java/org/apache/flink/streaming/connectors/redis/table/RedisSinkFunction.java

@@ -23,6 +23,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.time.LocalTime;
 import java.util.List;
 import java.util.Objects;
 
@@ -34,6 +35,7 @@ public class RedisSinkFunction<IN> extends RichSinkFunction<IN> {
  private static final Logger LOG = LoggerFactory.getLogger(RedisSinkFunction.class);
 
  protected Integer ttl;
+ protected int expireTimeSeconds = -1;
 
  private RedisSinkMapper<IN> redisSinkMapper;
  private RedisCommand redisCommand;
@@ -60,18 +62,21 @@ public RedisSinkFunction(
  ResolvedSchema resolvedSchema) {
  Objects.requireNonNull(flinkConfigBase, "Redis connection pool config should not be null");
  Objects.requireNonNull(redisSinkMapper, "Redis Mapper can not be null");
- Objects.requireNonNull(
- redisSinkMapper.getCommandDescription(),
- "Redis Mapper data type description can not be null");
 
  this.flinkConfigBase = flinkConfigBase;
  this.maxRetryTimes = redisSinkOptions.getMaxRetryTimes();
  this.redisSinkMapper = redisSinkMapper;
  RedisCommandDescription redisCommandDescription =
  (RedisCommandDescription) redisSinkMapper.getCommandDescription();
+ Preconditions.checkNotNull(
+ redisCommandDescription, "Redis Mapper data type description can not be null");
 
  this.redisCommand = redisCommandDescription.getRedisCommand();
  this.ttl = redisCommandDescription.getTTL();
+ if (redisCommandDescription.getExpireTime() != null) {
+ this.expireTimeSeconds = redisCommandDescription.getExpireTime().toSecondOfDay();
+ }
+
  this.columnDataTypes = resolvedSchema.getColumnDataTypes();
  this.redisValueDataStructure = redisSinkOptions.getRedisValueDataStructure();
  if (redisValueDataStructure == RedisValueDataStructure.row) {
@@ -123,9 +128,7 @@ private void startSink(String[] params) throws Exception {
  for (int i = 0; i <= maxRetryTimes; i++) {
  try {
  sink(params);
- if (ttl != null) {
- this.redisCommandsContainer.expire(params[0], ttl);
- }
+ setTtl(params[0]);
  break;
  } catch (UnsupportedOperationException e) {
  throw e;
@@ -207,6 +210,26 @@ private void sink(String[] params) {
  }
  }
 
+ /**
+ * set ttl for key.
+ *
+ * @param key
+ */
+ private void setTtl(String key) throws Exception {
+ if (expireTimeSeconds != -1) {
+ if (this.redisCommandsContainer.getTTL(key).get() == -1) {
+ int now = LocalTime.now().toSecondOfDay();
+ this.redisCommandsContainer.expire(
+ key,
+ expireTimeSeconds > now
+ ? expireTimeSeconds - now
+ : 86400 + expireTimeSeconds - now);
+ }
+ } else if (ttl != null) {
+ this.redisCommandsContainer.expire(key, ttl);
+ }
+ }
+
  /**
  * serialize whole row.
  *

src/test/java/org/apache/flink/streaming/connectors/redis/table/SQLTest.java

@@ -10,6 +10,8 @@
 import org.junit.jupiter.api.Test;
 import org.junit.platform.commons.util.Preconditions;
 
+import java.time.LocalTime;
+
 import static org.apache.flink.streaming.connectors.redis.descriptor.RedisValidator.REDIS_COMMAND;
 
 /** Created by jeff.zou on 2020/9/10. */
@@ -487,4 +489,82 @@ public void testMultiFieldLeftJoinForMap() throws Exception {
  Preconditions.condition(singleRedisCommands.hget("11_11", "11").equals("10.3"), "");
  Preconditions.condition(singleRedisCommands.hget("12_12", "12") == "", "");
  }
+
+ @Test
+ public void testHgetWithUpdate() throws Exception {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ EnvironmentSettings environmentSettings =
+ EnvironmentSettings.newInstance().inStreamingMode().build();
+ StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, environmentSettings);
+
+ clusterCommands.hset("key1", "field1", "1");
+ String dim =
+ "create table dim_table(name varchar, level varchar, age varchar) with ( 'connector'='redis', "
+ + "'cluster-nodes'='"
+ + CLUSTERNODES
+ + "','redis-mode'='cluster', 'password'='"
+ + CLUSTER_PASSWORD
+ + "','"
+ + REDIS_COMMAND
+ + "'='"
+ + RedisCommand.HGET
+ + "' )";
+
+ String source =
+ "create table source_table(age varchar, proctime as procTime()) "
+ + "with ('connector'='datagen', 'rows-per-second'='1', "
+ + "'fields.age.kind'='sequence', 'fields.age.start'='1', 'fields.age.end'='99'"
+ + ")";
+
+ String sink =
+ "create table sink_table(username varchar, level varchar,age varchar) with ( 'connector'='print')";
+
+ tEnv.executeSql(source);
+ tEnv.executeSql(dim);
+ tEnv.executeSql(sink);
+
+ // tEnv.executeSql("insert into dim_table select 'key1', 'field1', age from
+ // source_table ");
+ String sql =
+ " insert into sink_table "
+ + " select d.name, d.level, d.age from source_table s"
+ + " left join dim_table for system_time as of s.proctime as d "
+ + " on d.name ='key1' and d.level ='field1'";
+ TableResult tableResult = tEnv.executeSql(sql);
+ tableResult.getJobClient().get().getJobExecutionResult().get();
+ System.out.println(sql);
+ }
+
+ @Test
+ public void testSinkValueWithExpire() throws Exception {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
+
+ LocalTime localTime = LocalTime.now();
+ int wait = 100;
+ localTime = localTime.plusSeconds(wait);
+ String dim =
+ "create table sink_redis(name varchar, level varchar, age varchar) with ( 'connector'='redis', "
+ + "'cluster-nodes'='"
+ + CLUSTERNODES
+ + "','redis-mode'='cluster', 'ttl'='10','expire.on.time'='"
+ + localTime.toString()
+ + "', 'password'='"
+ + CLUSTER_PASSWORD
+ + "','"
+ + REDIS_COMMAND
+ + "'='"
+ + RedisCommand.HSET
+ + "' )";
+
+ tEnv.executeSql(dim);
+ String sql = " insert into sink_redis select * from (values ('1', '11.3', '10.3'))";
+ TableResult tableResult = tEnv.executeSql(sql);
+ tableResult.getJobClient().get().getJobExecutionResult().get();
+ System.out.println(sql);
+ Preconditions.condition(clusterCommands.exists("1") == 1, "");
+ Thread.sleep(wait * 1000);
+ Preconditions.condition(clusterCommands.exists("1") == 0, "");
+ }
 }