Added key reader to source connector. Resolves #3

Author: github-actions[bot]

src/docs/asciidoc/source.adoc

@@ -8,27 +8,56 @@ The {name} reads data from a Redis Enterprise database and publishes messages to
 The {name} includes the following features:
 
 * <<at-least-once-delivery,At least once delivery>>
+* <<tasks,Multiple tasks>>
+* <<key-reader,Key Reader>>
 * <<stream-reader,Stream Reader>>
 
 [[at-least-once-delivery]]
 === At least once delivery
 The {name} guarantees that records from the Kafka topic are delivered at least once.
 
+[[tasks]]
+=== Multiple Tasks
+Use configuration property `tasks.max` to have the change stream handled by multiple tasks. The connector splits the work based on the number of configured key patterns. When the number of tasks is greater than the number of patterns, the number of patterns will be used instead.
+
+[[key-reader]]
+=== Key Reader
+In key reader mode, the {name} captures changes happening to keys in a Redis database and publishes keys and values to a Kafka topic. The data structure key will be mapped to the record key, and the value will be mapped to the record value.
+
+[IMPORTANT]
+.Supported Data Structures
+====
+The {name} supports the following data structures:
+
+* String: the Kafka record values will be strings
+* Hash: the Kafka record values will be maps (string key/value pairs)
+
+====
+
+[source,properties]
+----
+redis.keys.patterns=<glob> # <1>
+topic=<topic> # <2>
+----
+
+<1> Key portion of the pattern that will be used to listen to keyspace events. For example `foo:*` translates to pubsub channel `$$__$$keyspace@0$$__$$:foo:*` and will capture changes to keys `foo:1`, `foo:2`, etc. Use comma-separated values for multiple patterns (`foo:*,bar:*`)
+<2> Name of the destination topic.
+
 [[stream-reader]]
 === Stream Reader
 In stream reader mode, the {name} reads messages from a Redis stream and publishes to a Kafka topic.
 
 [source,properties]
 ----
-redis.stream.name=<stream name> # <1>
-redis.stream.offset=<stream offset> # <2>
-redis.stream.count=<count> # <3>
+redis.reader=STREAM # <1>
+redis.stream.name=<stream name> # <2>
+redis.stream.offset=<stream offset> # <3>
 redis.stream.block=<millis> # <4>
-topic.name=<topic> # <5>
+topic=<topic> # <5>
 ----
 
-<1> Name of the stream to read from
-<2> https://redis.io/commands/xread#incomplete-ids[Message ID] to start reading from (default: `0-0`)
-<3> Maximum number of stream messages to include in a single https://redis.io/commands/xread[XREAD] when polling for new data (default: `50`). This setting can be used to limit the amount of data buffered internally in the connector
-<4> Maximum duration in milliseconds to wait while https://redis.io/commands/xread[reading] from the stream (default: 100)
-<5> A format string for the destination topic name (default: `${stream}`), which may contain `${stream}` as a placeholder for the originating stream name. For example, `redis_${stream}` for the stream 'orders' will map to the topic name `redis_orders`.
+<1> Reader mode is STREAM
+<2> Name of the stream to read from.
+<3> https://redis.io/commands/xread#incomplete-ids[Message ID] to start reading from (default: `0-0`).
+<4> Maximum duration in milliseconds to wait while https://redis.io/commands/xread[reading] from the stream (default: 100).
+<5> Name of the destination topic which may contain `${stream}` as a placeholder for the originating stream name. For example, `redis_${stream}` for the stream 'orders' will map to the topic name `redis_orders`.

src/main/java/com/redislabs/kafka/connect/RedisEnterpriseSourceConnector.java

@@ -15,21 +15,32 @@
 import com.redislabs.kafka.connect.source.RedisEnterpriseSourceConfig;
 import com.redislabs.kafka.connect.source.RedisEnterpriseSourceTask;
 import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigException;
 import org.apache.kafka.common.utils.AppInfoParser;
 import org.apache.kafka.connect.connector.Task;
+import org.apache.kafka.connect.errors.ConnectException;
 import org.apache.kafka.connect.source.SourceConnector;
+import org.apache.kafka.connect.util.ConnectorUtils;
 
-import java.util.Collections;
+import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 public class RedisEnterpriseSourceConnector extends SourceConnector {
 
     private Map<String, String> props;
+    private RedisEnterpriseSourceConfig config;
 
     @Override
     public void start(Map<String, String> props) {
         this.props = props;
+        try {
+            this.config = new RedisEnterpriseSourceConfig(props);
+        } catch (ConfigException configException) {
+            throw new ConnectException(configException);
+        }
     }
 
     @Override
@@ -39,7 +50,27 @@ public Class<? extends Task> taskClass() {
 
     @Override
     public List<Map<String, String>> taskConfigs(int maxTasks) {
-        return Collections.singletonList(props);
+        if (this.config.getReaderType() == RedisEnterpriseSourceConfig.ReaderType.KEYS) {
+            // Partition the configs based on channels
+            final List<List<String>> partitionedPatterns = ConnectorUtils
+                    .groupPartitions(this.config.getKeyPatterns(), Math.min(this.config.getKeyPatterns().size(), maxTasks));
+
+            // Create task configs based on the partitions
+            return partitionedPatterns.stream().map(this::taskConfig).collect(Collectors.toList());
+        }
+        List<Map<String, String>> taskConfigs = new ArrayList<>();
+        for (int i = 0; i < maxTasks; i++) {
+            Map<String, String> taskConfig = new HashMap<>(this.props);
+            taskConfig.put(RedisEnterpriseSourceTask.TASK_ID, Integer.toString(i));
+            taskConfigs.add(taskConfig);
+        }
+        return taskConfigs;
+    }
+
+    private Map<String, String> taskConfig(List<String> patterns) {
+        final Map<String, String> taskConfig = new HashMap<>(this.config.originalsStrings());
+        taskConfig.put(RedisEnterpriseSourceConfig.KEY_PATTERNS, String.join(",", patterns));
+        return taskConfig;
     }
 
     @Override
@@ -51,7 +82,6 @@ public ConfigDef config() {
         return new RedisEnterpriseSourceConfig.RedisEnterpriseSourceConfigDef();
     }
 
-
     @Override
     public String version() {
         return AppInfoParser.getVersion();

src/main/java/com/redislabs/kafka/connect/sink/RedisEnterpriseSinkTask.java

@@ -17,8 +17,9 @@
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.github.jcustenborder.kafka.connect.utils.data.SinkOffsetState;
-import com.github.jcustenborder.kafka.connect.utils.data.TopicPartitionCounter;
 import com.github.jcustenborder.kafka.connect.utils.jackson.ObjectMapperFactory;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
 import com.redislabs.kafka.connect.RedisEnterpriseSinkConnector;
 import io.lettuce.core.KeyValue;
 import io.lettuce.core.RedisClient;
@@ -55,6 +56,8 @@
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
 
 public class RedisEnterpriseSinkTask extends SinkTask {
 
@@ -144,10 +147,10 @@ private Double score(SinkRecord record) {
         if (value == null) {
             return null;
         }
-        if (value instanceof Double) {
-            return (Double) value;
+        if (value instanceof Number) {
+            return ((Number) value).doubleValue();
         }
-        throw new DataException("The value for the record must be float64 (Java double). Consider using a single message transformation to transform the data before it is written to Redis.");
+        throw new DataException("The value for the record must be a number. Consider using a single message transformation to transform the data before it is written to Redis.");
     }
 
     private boolean isDelete(SinkRecord record) {
@@ -224,11 +227,18 @@ public void put(final Collection<SinkRecord> records) {
         } catch (Exception e) {
             log.warn("Could not write {} records", records.size(), e);
         }
-        TopicPartitionCounter counter = new TopicPartitionCounter();
+        Map<TopicPartition, Long> data = new ConcurrentHashMap<>(100);
         for (SinkRecord record : records) {
-            counter.increment(record.topic(), record.kafkaPartition(), record.kafkaOffset());
+            Preconditions.checkState(!Strings.isNullOrEmpty(record.topic()), "topic cannot be null or empty.");
+            Preconditions.checkNotNull(record.kafkaPartition(), "partition cannot be null.");
+            Preconditions.checkState(record.kafkaOffset() >= 0, "offset must be greater than or equal 0.");
+            TopicPartition partition = new TopicPartition(record.topic(), record.kafkaPartition());
+            long current = data.getOrDefault(partition, Long.MIN_VALUE);
+            if (record.kafkaOffset() > current) {
+                data.put(partition, record.kafkaOffset());
+            }
         }
-        List<SinkOffsetState> offsetData = counter.offsetStates();
+        List<SinkOffsetState> offsetData = data.entrySet().stream().map(e -> SinkOffsetState.of(e.getKey(), e.getValue())).collect(Collectors.toList());
         if (!offsetData.isEmpty()) {
             Map<String, String> offsets = new LinkedHashMap<>(offsetData.size());
             for (SinkOffsetState e : offsetData) {

src/main/java/com/redislabs/kafka/connect/source/AbstractSourceRecordReader.java

@@ -0,0 +1,57 @@
+package com.redislabs.kafka.connect.source;
+
+import com.redislabs.mesclun.RedisModulesClient;
+import io.lettuce.core.RedisClient;
+import org.apache.kafka.connect.source.SourceRecord;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.util.Assert;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+public abstract class AbstractSourceRecordReader<T> implements SourceRecordReader {
+
+    private static final Logger log = LoggerFactory.getLogger(AbstractSourceRecordReader.class);
+
+    protected final RedisEnterpriseSourceConfig sourceConfig;
+    private RedisClient client;
+
+    protected AbstractSourceRecordReader(RedisEnterpriseSourceConfig sourceConfig) {
+        Assert.notNull(sourceConfig, "Source connector config must not be null");
+        this.sourceConfig = sourceConfig;
+    }
+
+    @Override
+    public void open() {
+        this.client = RedisModulesClient.create(sourceConfig.getRedisUri());
+        open(client);
+    }
+
+    protected abstract void open(RedisClient client);
+
+    @Override
+    public List<SourceRecord> poll() {
+        List<T> records;
+        try {
+            records = doPoll();
+        } catch (Exception e) {
+            log.error("Could not read messages", e);
+            return null;
+        }
+        return records.stream().map(this::convert).collect(Collectors.toList());
+    }
+
+    protected abstract List<T> doPoll() throws Exception;
+
+    protected abstract SourceRecord convert(T input);
+
+    @Override
+    public void close() {
+        doClose();
+        client.shutdown();
+        client.getResources().shutdown();
+    }
+
+    protected abstract void doClose();
+}

src/main/java/com/redislabs/kafka/connect/source/KeySourceRecordReader.java

@@ -0,0 +1,68 @@
+package com.redislabs.kafka.connect.source;
+
+import io.lettuce.core.RedisClient;
+import org.apache.kafka.connect.data.Schema;
+import org.apache.kafka.connect.data.SchemaBuilder;
+import org.apache.kafka.connect.source.SourceRecord;
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.batch.item.redis.DataStructureItemReader;
+import org.springframework.batch.item.redis.support.DataStructure;
+import org.springframework.batch.item.redis.support.LiveKeyValueItemReader;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class KeySourceRecordReader extends AbstractSourceRecordReader<DataStructure> {
+
+    private static final Schema KEY_SCHEMA = Schema.STRING_SCHEMA;
+    private static final Schema STRING_VALUE_SCHEMA = Schema.STRING_SCHEMA;
+    private static final String HASH_VALUE_SCHEMA_NAME = "com.redislabs.kafka.connect.HashEventValue";
+    private static final Schema HASH_VALUE_SCHEMA = SchemaBuilder.map(Schema.STRING_SCHEMA, Schema.STRING_SCHEMA).name(HASH_VALUE_SCHEMA_NAME);
+
+    private final int batchSize;
+    private final String topic;
+    private LiveKeyValueItemReader<DataStructure> reader;
+    private final Duration idleTimeout;
+
+    public KeySourceRecordReader(RedisEnterpriseSourceConfig sourceConfig, Duration idleTimeout) {
+        super(sourceConfig);
+        this.topic = sourceConfig.getTopicName();
+        this.batchSize = Math.toIntExact(sourceConfig.getBatchSize());
+        this.idleTimeout = idleTimeout;
+    }
+
+    @Override
+    protected void open(RedisClient client) {
+        reader = DataStructureItemReader.client(client).live().idleTimeout(idleTimeout).keyPatterns(sourceConfig.getKeyPatterns().toArray(new String[0])).build();
+        reader.open(new ExecutionContext());
+    }
+
+    @Override
+    protected List<DataStructure> doPoll() throws Exception {
+        return reader.read(batchSize);
+    }
+
+    @Override
+    protected SourceRecord convert(DataStructure input) {
+        Map<String, ?> sourcePartition = new HashMap<>();
+        Map<String, ?> sourceOffset = new HashMap<>();
+        return new SourceRecord(sourcePartition, sourceOffset, topic, null, KEY_SCHEMA, input.getKey(), schema(input), input.getValue(), Instant.now().getEpochSecond());
+    }
+
+    private Schema schema(DataStructure input) {
+        if (DataStructure.HASH.equals(input.getType())) {
+            return HASH_VALUE_SCHEMA;
+        }
+        return STRING_VALUE_SCHEMA;
+    }
+
+    @Override
+    protected void doClose() {
+        if (reader != null) {
+            reader.close();
+        }
+    }
+}