feat(source): Added mode config for LIVE or LIVEONLY key reader

Author: jruaux

core/redis-kafka-connect/src/main/java/com/redis/kafka/connect/source/RedisKeysSourceConfig.java

@@ -15,6 +15,8 @@
  */
 package com.redis.kafka.connect.source;
 
+import com.redis.spring.batch.item.redis.RedisItemReader;
+
 import java.time.Duration;
 import java.util.Map;
 import java.util.Objects;
@@ -27,13 +29,16 @@ public class RedisKeysSourceConfig extends RedisSourceConfig {
 
     private final String topicName;
 
-    private Duration idleTimeout;
+    private final Duration idleTimeout;
+
+    private final RedisItemReader.ReaderMode mode;
 
     public RedisKeysSourceConfig(Map<?, ?> originals) {
         super(new RedisKeysSourceConfigDef(), originals);
         this.topicName = getString(RedisKeysSourceConfigDef.TOPIC_CONFIG);
         this.keyPattern = getString(RedisKeysSourceConfigDef.KEY_PATTERN_CONFIG);
         this.idleTimeout = Duration.ofMillis(getLong(RedisKeysSourceConfigDef.IDLE_TIMEOUT_CONFIG));
+        this.mode = RedisItemReader.ReaderMode.valueOf(getString(RedisKeysSourceConfigDef.MODE_CONFIG).toUpperCase());
     }
 
     public String getKeyPattern() {
@@ -48,6 +53,10 @@ public Duration getIdleTimeout() {
         return idleTimeout;
     }
 
+    public RedisItemReader.ReaderMode getMode() {
+        return mode;
+    }
+
     @Override
     public int hashCode() {
         final int prime = 31;

core/redis-kafka-connect/src/main/java/com/redis/kafka/connect/source/RedisKeysSourceConfigDef.java

@@ -1,5 +1,7 @@
 package com.redis.kafka.connect.source;
 
+import com.redis.spring.batch.item.redis.RedisItemReader;
+
 public class RedisKeysSourceConfigDef extends RedisSourceConfigDef {
 
 	public static final String TOPIC_CONFIG = "topic";
@@ -14,6 +16,10 @@ public class RedisKeysSourceConfigDef extends RedisSourceConfigDef {
 	public static final String IDLE_TIMEOUT_DEFAULT = "0";
 	public static final String IDLE_TIMEOUT_DOC = "Idle timeout in millis. Use 0 to disable.";
 
+	public static final String MODE_CONFIG = "mode";
+	public static final RedisItemReader.ReaderMode MODE_DEFAULT = RedisItemReader.ReaderMode.LIVE;
+	public static final String MODE_DOC = "Key reading mode. Use LIVE for snapshot + updates, LIVEONLY for just updates.";
+
 	public RedisKeysSourceConfigDef() {
 		define();
 	}
@@ -27,6 +33,7 @@ private void define() {
 		define(TOPIC_CONFIG, Type.STRING, TOPIC_DEFAULT, Importance.MEDIUM, TOPIC_DOC);
 		define(KEY_PATTERN_CONFIG, Type.STRING, KEY_PATTERN_DEFAULT, Importance.MEDIUM, KEY_PATTERN_DOC);
 		define(IDLE_TIMEOUT_CONFIG, Type.LONG, IDLE_TIMEOUT_DEFAULT, Importance.LOW, IDLE_TIMEOUT_DOC);
+		define(MODE_CONFIG, Type.STRING, MODE_DEFAULT.name(), Importance.MEDIUM, MODE_DOC);
 	}
 
-}
+}

core/redis-kafka-connect/src/main/java/com/redis/kafka/connect/source/RedisKeysSourceTask.java

@@ -15,13 +15,10 @@
  */
 package com.redis.kafka.connect.source;
 
-import java.time.Clock;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.UUID;
-
+import com.redis.kafka.connect.common.ManifestVersionProvider;
+import com.redis.spring.batch.item.redis.RedisItemReader;
+import com.redis.spring.batch.item.redis.common.KeyValue;
+import io.lettuce.core.AbstractRedisClient;
 import org.apache.kafka.connect.data.Schema;
 import org.apache.kafka.connect.errors.ConnectException;
 import org.apache.kafka.connect.errors.RetriableException;
@@ -30,12 +27,8 @@
 import org.springframework.batch.item.ExecutionContext;
 import org.springframework.batch.item.ItemStreamException;
 
-import com.redis.kafka.connect.common.ManifestVersionProvider;
-import com.redis.spring.batch.item.redis.RedisItemReader;
-import com.redis.spring.batch.item.redis.RedisItemReader.ReaderMode;
-import com.redis.spring.batch.item.redis.common.KeyValue;
-
-import io.lettuce.core.AbstractRedisClient;
+import java.time.Clock;
+import java.util.*;
 
 public class RedisKeysSourceTask extends SourceTask {
 
@@ -76,7 +69,7 @@ public void start(Map<String, String> props) {
 		// Use a random job name to not interfere with other key source tasks
 		reader.setName(UUID.randomUUID().toString());
 		reader.setClient(client);
-		reader.setMode(ReaderMode.LIVE);
+		reader.setMode(config.getMode());
 		reader.setPoolSize(config.getPoolSize());
 		reader.setDatabase(config.uri().getDatabase());
 		reader.setKeyPattern(config.getKeyPattern());