Added keys reader to source connector. Resolves #2

Author: jruaux

pom.xml

@@ -62,9 +62,9 @@
     </repositories>
     <dependencies>
         <dependency>
-            <groupId>io.lettuce</groupId>
-            <artifactId>lettuce-core</artifactId>
-            <version>6.1.3.RELEASE</version>
+            <groupId>com.redislabs</groupId>
+            <artifactId>mesclun</artifactId>
+            <version>1.3.4</version>
         </dependency>
         <dependency>
             <groupId>org.slf4j</groupId>

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

@@ -59,6 +59,7 @@
 public class RedisEnterpriseSinkTask extends SinkTask {
 
     private static final Logger log = LoggerFactory.getLogger(RedisEnterpriseSinkTask.class);
+    private static final String OFFSET_KEY_FORMAT = "com.redislabs.kafka.connect.sink.offset.%s.%s";
 
     private RedisClient client;
     private RedisEnterpriseSinkConfig config;
@@ -82,7 +83,7 @@ public void start(final Map<String, String> props) {
         final java.util.Set<TopicPartition> assignment = this.context.assignment();
         if (!assignment.isEmpty()) {
             List<SinkOffsetState> offsetStates = new ArrayList<>();
-            String[] partitionKeys = assignment.stream().map(this::offsetKey).toArray(String[]::new);
+            String[] partitionKeys = assignment.stream().map(a -> offsetKey(a.topic(), a.partition())).toArray(String[]::new);
             List<KeyValue<String, String>> values = connection.sync().mget(partitionKeys);
             for (KeyValue<String, String> value : values) {
                 if (value.hasValue()) {
@@ -108,8 +109,8 @@ public void start(final Map<String, String> props) {
         }
     }
 
-    private String offsetKey(TopicPartition topicPartition) {
-        return String.format("__kafka.offset.%s.%s", topicPartition.topic(), topicPartition.partition());
+    private String offsetKey(String topic, Integer partition) {
+        return String.format(OFFSET_KEY_FORMAT, topic, partition);
     }
 
     private OperationItemWriter.RedisOperation<byte[], byte[], SinkRecord> operation() {
@@ -231,7 +232,7 @@ public void put(final Collection<SinkRecord> records) {
         if (!offsetData.isEmpty()) {
             Map<String, String> offsets = new LinkedHashMap<>(offsetData.size());
             for (SinkOffsetState e : offsetData) {
-                String key = String.format("__kafka.offset.%s.%s", e.topic(), e.partition());
+                String key = offsetKey(e.topic(), e.partition());
                 String value;
                 try {
                     value = ObjectMapperFactory.INSTANCE.writeValueAsString(e);

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

@@ -16,16 +16,35 @@
 package com.redislabs.kafka.connect.source;
 
 import com.github.jcustenborder.kafka.connect.utils.config.ConfigKeyBuilder;
+import com.github.jcustenborder.kafka.connect.utils.config.ConfigUtils;
+import com.github.jcustenborder.kafka.connect.utils.config.validators.Validators;
 import com.redislabs.kafka.connect.common.RedisEnterpriseConfig;
 import org.apache.kafka.common.config.ConfigDef;
 
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 
 public class RedisEnterpriseSourceConfig extends RedisEnterpriseConfig {
 
+
     public static final String TOKEN_STREAM = "${stream}";
 
+    public static final String READER_TYPE = "redis.reader";
+    public static final String READER_DEFAULT = ReaderType.KEYS.name();
+    public static final String READER_DOC = "Source from which to read Redis records. " + ReaderType.KEYS + ": generate records from key events and respective values generated from write operations in the Redis database. " + ReaderType.STREAM + ": read messages from a Redis stream";
+
+    public static final String KEYS_PATTERN = "redis.keys.pattern";
+    public static final String KEYS_PATTERN_DEFAULT = "*";
+    public static final String KEYS_PATTERN_DOC = "Generate records only for key names that match the given glob-style pattern";
+
+    public static final String KEYS_EVENT_TYPES = "redis.keys.eventtypes";
+    public static final String KEYS_EVENT_TYPES_DEFAULT = "";
+    public static final String KEYS_EVENT_TYPES_DOC = "Comma-separated event types for key reader. The list may contain one or more command or event (default: None)";
+
     public static final String STREAM_NAME = "redis.stream.name";
+    public static final String STREAM_NAME_DEFAULT = "com.redislabs.kafka.connect.source.stream";
     public static final String STREAM_NAME_DOC = "Name of the Redis stream to read from";
 
     public static final String STREAM_OFFSET = "redis.stream.offset";
@@ -42,8 +61,11 @@ public class RedisEnterpriseSourceConfig extends RedisEnterpriseConfig {
 
     public static final String TOPIC_NAME_FORMAT = "topic.name";
     public static final String TOPIC_NAME_FORMAT_DEFAULT = TOKEN_STREAM;
-    public static final String TOPIC_NAME_FORMAT_DOC = "A format string for the destination topic name, which may contain '${stream}' as a " + "placeholder for the originating stream name.\n" + "For example, ``redis_${stream}`` for the stream 'orders' will map to the topic name " + "'redis_orders'.";
+    public static final String TOPIC_NAME_FORMAT_DOC = String.format("A format string for the destination topic name, which may contain '%s' as a placeholder for the originating stream name. For example `redis_%s` for the stream 'orders' will map to the topic name 'redis_orders'.", TOKEN_STREAM, TOKEN_STREAM);
 
+    private final ReaderType readerType;
+    private final List<String> keysEventTypes;
+    private final String keysPattern;
     private final String streamName;
     private final String streamOffset;
     private final Long streamCount;
@@ -52,13 +74,35 @@ public class RedisEnterpriseSourceConfig extends RedisEnterpriseConfig {
 
     public RedisEnterpriseSourceConfig(Map<?, ?> originals) {
         super(new RedisEnterpriseSourceConfigDef(), originals);
+        this.readerType = ConfigUtils.getEnum(ReaderType.class, this, READER_TYPE);
+        this.keysEventTypes = splitCommaSeparated(getString(KEYS_EVENT_TYPES));
+        this.keysPattern = getString(KEYS_PATTERN);
         this.streamName = getString(STREAM_NAME);
         this.streamOffset = getString(STREAM_OFFSET);
         this.streamCount = getLong(STREAM_COUNT);
         this.streamBlock = getLong(STREAM_BLOCK);
         this.topicNameFormat = getString(TOPIC_NAME_FORMAT);
     }
 
+    private List<String> splitCommaSeparated(String string) {
+        if (string.trim().isEmpty()) {
+            return Collections.emptyList();
+        }
+        return Arrays.asList(string.split(","));
+    }
+
+    public List<String> getKeysEventTypes() {
+        return keysEventTypes;
+    }
+
+    public ReaderType getReaderType() {
+        return readerType;
+    }
+
+    public String getKeysPattern() {
+        return keysPattern;
+    }
+
     public Long getStreamBlock() {
         return streamBlock;
     }
@@ -91,14 +135,20 @@ public RedisEnterpriseSourceConfigDef(ConfigDef base) {
         }
 
         private void define() {
-            define(ConfigKeyBuilder.of(STREAM_NAME, ConfigDef.Type.STRING).documentation(STREAM_NAME_DOC).importance(ConfigDef.Importance.HIGH).validator(ConfigDef.CompositeValidator.of(new ConfigDef.NonNullValidator(), new ConfigDef.NonEmptyString())).build());
+            define(ConfigKeyBuilder.of(READER_TYPE, ConfigDef.Type.STRING).documentation(READER_DOC).defaultValue(READER_DEFAULT).importance(ConfigDef.Importance.HIGH).validator(Validators.validEnum(ReaderType.class)).build());
+            define(ConfigKeyBuilder.of(KEYS_EVENT_TYPES, Type.STRING).documentation(KEYS_EVENT_TYPES_DOC).defaultValue(KEYS_EVENT_TYPES_DEFAULT).importance(Importance.MEDIUM).build());
+            define(ConfigKeyBuilder.of(KEYS_PATTERN, Type.STRING).documentation(KEYS_PATTERN_DOC).defaultValue(KEYS_PATTERN_DEFAULT).importance(Importance.MEDIUM).build());
+            define(ConfigKeyBuilder.of(STREAM_NAME, ConfigDef.Type.STRING).documentation(STREAM_NAME_DOC).defaultValue(STREAM_NAME_DEFAULT).importance(ConfigDef.Importance.HIGH).build());
             define(ConfigKeyBuilder.of(STREAM_OFFSET, ConfigDef.Type.STRING).documentation(STREAM_OFFSET_DOC).defaultValue(STREAM_OFFSET_DEFAULT).importance(ConfigDef.Importance.MEDIUM).build());
             define(ConfigKeyBuilder.of(STREAM_COUNT, ConfigDef.Type.LONG).defaultValue(STREAM_COUNT_DEFAULT).importance(ConfigDef.Importance.LOW).documentation(STREAM_COUNT_DOC).validator(ConfigDef.Range.atLeast(1L)).build());
             define(ConfigKeyBuilder.of(STREAM_BLOCK, ConfigDef.Type.LONG).defaultValue(STREAM_BLOCK_DEFAULT).importance(ConfigDef.Importance.LOW).documentation(STREAM_BLOCK_DOC).validator(ConfigDef.Range.atLeast(1L)).build());
             define(ConfigKeyBuilder.of(TOPIC_NAME_FORMAT, ConfigDef.Type.STRING).defaultValue(TOPIC_NAME_FORMAT_DEFAULT).importance(ConfigDef.Importance.MEDIUM).documentation(TOPIC_NAME_FORMAT_DOC).build());
         }
 
+    }
 
+    public enum ReaderType {
+        KEYS, STREAM
     }
 
 }

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

@@ -15,8 +15,12 @@
  */
 package com.redislabs.kafka.connect.source;
 
+import com.google.common.base.Charsets;
 import com.redislabs.kafka.connect.RedisEnterpriseSourceConnector;
-import io.lettuce.core.RedisClient;
+import com.redislabs.mesclun.RedisModulesClient;
+import com.redislabs.mesclun.api.StatefulRedisModulesConnection;
+import com.redislabs.mesclun.api.sync.RedisGearsCommands;
+import com.redislabs.mesclun.gears.RedisGearsUtils;
 import io.lettuce.core.StreamMessage;
 import io.lettuce.core.XReadArgs;
 import org.apache.kafka.connect.data.Schema;
@@ -35,19 +39,23 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 public class RedisEnterpriseSourceTask extends SourceTask {
 
     private static final Logger log = LoggerFactory.getLogger(RedisEnterpriseSourceTask.class);
 
-    public static final String STREAM_FIELD = "stream";
+    public static final String OFFSET = "stream";
     public static final String OFFSET_FIELD = "offset";
-
+    private static final String KEYSREADER_STREAM = "${stream}";
+    private static final String KEYSREADER_PATTERN = "${prefix}";
+    private static final String KEYSREADER_EVENTTYPES = "${eventTypes}";
+    private static final String EVENTTYPES_NONE = "None";
     private static final Schema KEY_SCHEMA = Schema.STRING_SCHEMA;
     private static final String VALUE_SCHEMA_NAME = "com.redislabs.kafka.connect.EventValue";
     private static final Schema VALUE_SCHEMA = SchemaBuilder.map(Schema.STRING_SCHEMA, Schema.STRING_SCHEMA).name(VALUE_SCHEMA_NAME);
 
-    private RedisClient client;
+    private RedisModulesClient client;
     private RedisEnterpriseSourceConfig sourceConfig;
     private StreamItemReader reader;
     private Map<String, String> offsetKey;
@@ -60,8 +68,17 @@ public String version() {
     @Override
     public void start(Map<String, String> props) {
         this.sourceConfig = new RedisEnterpriseSourceConfig(props);
-        this.client = RedisClient.create(sourceConfig.getRedisUri());
-        this.offsetKey = Collections.singletonMap(STREAM_FIELD, sourceConfig.getStreamName());
+        this.client = RedisModulesClient.create(sourceConfig.getRedisUri());
+        if (sourceConfig.getReaderType() == RedisEnterpriseSourceConfig.ReaderType.KEYS) {
+            StatefulRedisModulesConnection<String, String> connection = client.connect();
+            RedisGearsCommands<String, String> sync = connection.sync();
+            String function = RedisGearsUtils.toString(getClass().getClassLoader().getResourceAsStream("keysreader.py"), Charsets.UTF_8);
+            function = function.replace(KEYSREADER_STREAM, sourceConfig.getStreamName());
+            function = function.replace(KEYSREADER_EVENTTYPES, toPyArray(sourceConfig.getKeysEventTypes()));
+            function = function.replace(KEYSREADER_PATTERN, sourceConfig.getKeysPattern());
+            sync.pyExecute(function);
+        }
+        this.offsetKey = Collections.singletonMap(OFFSET, sourceConfig.getStreamName());
         String offset = sourceConfig.getStreamOffset();
         if (context != null) {
             Map<String, Object> storedOffset = context.offsetStorageReader().offset(offsetKey);
@@ -82,6 +99,13 @@ public void start(Map<String, String> props) {
         this.reader.open(new ExecutionContext());
     }
 
+    private String toPyArray(List<String> eventTypes) {
+        if (eventTypes.isEmpty()) {
+            return EVENTTYPES_NONE;
+        }
+        return eventTypes.stream().map(s -> "'" + s + "'").collect(Collectors.joining(",", "[", "]"));
+    }
+
     @Override
     public void stop() {
         if (reader != null) {

src/main/resources/keysreader.py

@@ -0,0 +1,8 @@
+# Capture each keyspace event and store it in a stream
+GB('KeysReader') \
+.foreach(lambda x:
+            execute('XADD', '${stream}', '*', 'key', x['key'], 'value', x['value'], 'type', x['type'], 'event', x['event'])) \
+.register(prefix='${prefix}',
+             mode='sync',
+             eventTypes=${eventTypes},
+             readValue=True)

src/main/resources/gears/streamreader.py renamed to src/main/resources/streamreader.py

@@ -1,7 +1,9 @@
 package com.redislabs.kafka.connect;
 
 import com.redislabs.testcontainers.RedisEnterpriseContainer;
+import com.redislabs.testcontainers.RedisModulesContainer;
 import com.redislabs.testcontainers.RedisServer;
+import com.redislabs.testcontainers.support.enterprise.rest.Database;
 import io.lettuce.core.AbstractRedisClient;
 import io.lettuce.core.RedisClient;
 import io.lettuce.core.api.StatefulConnection;
@@ -36,7 +38,8 @@
 public class AbstractRedisEnterpriseIT {
 
     @Container
-    private static final RedisEnterpriseContainer REDIS_ENTERPRISE = new RedisEnterpriseContainer();
+    private static final RedisEnterpriseContainer REDIS_ENTERPRISE = new RedisEnterpriseContainer().withModules(Database.Module.GEARS);
+//    private static final RedisModulesContainer REDIS_ENTERPRISE = new RedisModulesContainer();
 
     static Stream<RedisServer> redisServers() {
         return Stream.of(REDIS_ENTERPRISE);

src/test/java/com/redislabs/kafka/connect/AbstractRedisEnterpriseIT.java

@@ -20,9 +20,7 @@ public void testConfig() {
         Map<String, ConfigValue> results = config.validateAll(new HashMap<>());
         ConfigValue value = results.get(RedisEnterpriseSourceConfig.STREAM_NAME);
         Assertions.assertEquals(RedisEnterpriseSourceConfig.STREAM_NAME, value.name());
-        Assertions.assertNull(value.value());
-        String expected = "Missing required configuration \"" + RedisEnterpriseSourceConfig.STREAM_NAME + "\" which has no default value.";
-        Assertions.assertEquals(expected, value.errorMessages().get(0));
+        Assertions.assertEquals(RedisEnterpriseSourceConfig.STREAM_NAME_DEFAULT, value.value());
     }
 
     @Test

src/test/java/com/redislabs/kafka/connect/RedisEnterpriseSourceConnectorTest.java

@@ -3,34 +3,34 @@
 import com.redislabs.kafka.connect.source.RedisEnterpriseSourceConfig;
 import com.redislabs.kafka.connect.source.RedisEnterpriseSourceTask;
 import com.redislabs.testcontainers.RedisServer;
+import io.lettuce.core.RedisClient;
+import io.lettuce.core.RedisURI;
+import io.lettuce.core.codec.ByteArrayCodec;
 import org.apache.kafka.connect.source.SourceRecord;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.MethodSource;
 
-import java.util.HashMap;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
 public class RedisEnterpriseSourceTaskIT extends AbstractRedisEnterpriseIT {
 
-    private static final String STREAM = "stream1";
-
     private RedisEnterpriseSourceTask task;
 
     @BeforeEach
     public void createTask() {
         task = new RedisEnterpriseSourceTask();
     }
 
-    private void startTask(RedisServer redis) {
-        final Map<String, String> config = new HashMap<>();
+    private void startTask(RedisServer redis, String... props) {
+        Map<String, String> config = map(props);
         config.put(RedisEnterpriseSourceConfig.REDIS_URI, redis.getRedisURI());
-        config.put(RedisEnterpriseSourceConfig.STREAM_NAME, STREAM);
         task.start(config);
     }
 
@@ -42,19 +42,53 @@ public void teardown() {
     @SuppressWarnings("unchecked")
     @ParameterizedTest
     @MethodSource("redisServers")
-    public void pollRetrievesStreamMessages(RedisServer redis) throws InterruptedException {
-        startTask(redis);
-        syncStream(redis).xadd(STREAM, map("field1", "value1", "field2", "value2"));
-        syncStream(redis).xadd(STREAM, map("field1", "value1", "field2", "value2"));
-        syncStream(redis).xadd(STREAM, map("field1", "value1", "field2", "value2"));
+    public void pollStream(RedisServer redis) throws InterruptedException {
+        final String stream = "stream1";
+        startTask(redis, RedisEnterpriseSourceConfig.READER_TYPE, RedisEnterpriseSourceConfig.ReaderType.STREAM.name(), RedisEnterpriseSourceConfig.STREAM_NAME, stream);
+        String field1 = "field1";
+        String value1 = "value1";
+        String field2 = "field2";
+        String value2 = "value2";
+        syncStream(redis).xadd(stream, map(field1, value1, field2, value2));
+        syncStream(redis).xadd(stream, map(field1, value1, field2, value2));
+        syncStream(redis).xadd(stream, map(field1, value1, field2, value2));
         Thread.sleep(100);
-        final List<SourceRecord> sourceRecords = task.poll();
+        List<SourceRecord> sourceRecords = task.poll();
         assertEquals(3, sourceRecords.size());
         for (SourceRecord record : sourceRecords) {
-            Assertions.assertEquals(STREAM, record.topic());
+            Assertions.assertEquals(stream, record.topic());
             Map<String, String> map = (Map<String, String>) record.value();
-            Assertions.assertEquals("value1", map.get("field1"));
-            Assertions.assertEquals("value2", map.get("field2"));
+            Assertions.assertEquals(value1, map.get(field1));
+            Assertions.assertEquals(value2, map.get(field2));
+        }
+    }
+
+    @SuppressWarnings("unchecked")
+    @ParameterizedTest
+    @MethodSource("redisServers")
+    public void pollKeys(RedisServer redis) throws InterruptedException {
+        startTask(redis);
+        Thread.sleep(1000);
+        String stringKey = "key:1";
+        String stringValue = "my string";
+        syncString(redis).set(stringKey, stringValue);
+        String hashKey = "key:2";
+        syncHash(redis).hset(hashKey, map("field1", "value1", "field2", "value2"));
+        Thread.sleep(1000);
+        List<SourceRecord> sourceRecords = task.poll();
+        assertEquals(2, sourceRecords.size());
+        SourceRecord stringRecord = sourceRecords.get(0);
+        Map<String, String> stringBody = (Map<String, String>) stringRecord.value();
+        assertEquals("string", stringBody.get("type"));
+        assertEquals(stringKey, stringBody.get("key"));
+        assertEquals(stringValue, stringBody.get("value"));
+    }
+
+    public static void main(String[] args) {
+        RedisClient client = RedisClient.create(RedisURI.create("localhost", 12000));
+        List<byte[]> keys = client.connect(new ByteArrayCodec()).sync().keys("*".getBytes());
+        for (byte[] key : keys) {
+            System.out.println(Arrays.hashCode(key));
         }
     }
 }