[source] Made stream the default reader and removed key reader entry from docs

Author: github-actions[bot]

docker/run.sh

@@ -103,7 +103,6 @@ curl -X POST -H "Content-Type: application/json" --data '
    "config": {
      "tasks.max":"1",
      "connector.class":"com.redislabs.kafka.connect.RedisEnterpriseSourceConnector",
-     "redis.reader":"STREAM",
      "redis.uri":"redis://redis:6379",
      "redis.stream.name":"mystream",
      "topic": "mystream"

src/docs/asciidoc/sink.adoc

@@ -103,7 +103,7 @@ value.converter.schema.registry.url=http://localhost:8081
 [source,properties]
 ----
 value.converter=org.apache.kafka.connect.json.JsonConverter
-value.converter.schemas.enable=<true|false> # <1>
+value.converter.schemas.enable=<true|false> <1>
 ----
 
 <1> Set to `true` if the JSON record structure has an attached schema
@@ -123,8 +123,8 @@ Use the following properties to store Kafka records as Redis stream messages:
 [source,properties]
 ----
 redis.type=STREAM
-redis.key=<stream key> # <1>
-value.converter=<Avro or JSON> # <2>
+redis.key=<stream key> <1>
+value.converter=<Avro or JSON> <2>
 ----
 
 <1> <<collection-key,Stream key>>
@@ -136,8 +136,8 @@ Use the following properties to write Kafka records as Redis hashes:
 [source,properties]
 ----
 redis.type=HASH
-key.converter=<string or bytes> # <1>
-value.converter=<Avro or JSON> # <2>
+key.converter=<string or bytes> <1>
+value.converter=<Avro or JSON> <2>
 ----
 
 <1> <<key-string,String>> or <<key-bytes,bytes>>
@@ -149,8 +149,8 @@ Use the following properties to write Kafka records as Redis strings:
 [source,properties]
 ----
 redis.type=STRING
-key.converter=<string or bytes> # <1>
-value.converter=<string or bytes> # <2>
+key.converter=<string or bytes> <1>
+value.converter=<string or bytes> <2>
 ----
 
 <1> <<key-string,String>> or <<key-bytes,bytes>>
@@ -162,9 +162,9 @@ Use the following properties to add Kafka record keys to a Redis list:
 [source,properties]
 ----
 redis.type=LIST
-redis.key=<key name> # <1>
-key.converter=<string or bytes> # <2>
-redis.push.direction=<LEFT or RIGHT> # <3>
+redis.key=<key name> <1>
+key.converter=<string or bytes> <2>
+redis.push.direction=<LEFT or RIGHT> <3>
 ----
 
 <1> <<collection-key,List key>>
@@ -179,8 +179,8 @@ Use the following properties to add Kafka record keys to a Redis set:
 [source,properties]
 ----
 redis.type=SET
-redis.key=<key name> # <1>
-key.converter=<string or bytes> # <2>
+redis.key=<key name> <1>
+key.converter=<string or bytes> <2>
 ----
 
 <1> <<collection-key,Set key>>
@@ -194,8 +194,8 @@ Use the following properties to add Kafka record keys to a Redis sorted set:
 [source,properties]
 ----
 redis.type=ZSET
-redis.key=<key name> # <1>
-key.converter=<string or bytes> # <2>
+redis.key=<key name> <1>
+key.converter=<string or bytes> <2>
 ----
 
 <1> <<collection-key,Sorted set key>>

src/docs/asciidoc/source.adoc

@@ -1,15 +1,14 @@
 = Source Connector Guide
 :name: Redis Enterprise Kafka Source Connector
 
-The {name} reads data from a Redis Enterprise database and publishes messages to a Kafka topic.
+The {name} reads from a Redis Enterprise stream and publishes messages to a Kafka topic.
 
 == Features
 
 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]]
@@ -20,44 +19,47 @@ The {name} guarantees that records from the Kafka topic are delivered at least o
 === 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.
+//
+//[[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.
+The {name} reads messages from a stream and publishes to a Kafka topic. Reading is done through a consumer group so that <<multiple-tasks,multiple instances>> of the connector configured via the `tasks.max` can consume messages in a round-robin fashion.
 
 [source,properties]
 ----
-redis.reader=STREAM # <1>
-redis.stream.name=<stream name> # <2>
-redis.stream.offset=<stream offset> # <3>
-redis.stream.block=<millis> # <4>
-topic=<topic> # <5>
+redis.stream.name=<name> <1>
+redis.stream.offset=<offset> <2>
+redis.stream.block=<millis> <3>
+redis.stream.consumer.group=<group> <4>
+redis.stream.consumer.name=<name> <5>
+topic=<name> <6>
 ----
 
-<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`.
+<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 https://redis.io/commands/xread[XREAD] wait duration in milliseconds (default: `100`).
+<4> Name of the stream consumer group (default: `kafka-consumer-group`).
+<5> Name of the stream consumer (default: `consumer-${task}`). May contain `${task}` as a placeholder for the task id. For example, `foo${task}` and task `123` => consumer `foo123`.
+<6> Destination topic (default: `${stream}`). May contain `${stream}` as a placeholder for the originating stream name. For example, `redis_${stream}` and stream `orders` => topic `redis_orders`.

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

@@ -21,20 +21,20 @@
 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 TOKEN_TASK = "${task}";
 
     public static final String TOPIC = "topic";
-    public static final String TOPIC_DOC = String.format("Name of the destination topic. In stream reader mode the topic name 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);
+    public static final String TOPIC_DEFAULT = TOKEN_STREAM;
+    public static final String TOPIC_DOC = String.format("Name of the destination topic, 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);
 
     public static final String READER_TYPE = "redis.reader";
-    public static final String READER_DEFAULT = ReaderType.KEYS.name();
+    public static final String READER_DEFAULT = ReaderType.STREAM.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 BATCH_SIZE = "batch.size";
@@ -46,7 +46,6 @@ public class RedisEnterpriseSourceConfig extends RedisEnterpriseConfig {
     public static final String KEY_PATTERNS_DOC = "Keyspace glob-style patterns to subscribe to, comma-separated.";
 
     public static final String STREAM_NAME = "redis.stream.name";
-    public static final String STREAM_NAME_DEFAULT = "kafka: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";
@@ -57,6 +56,11 @@ public class RedisEnterpriseSourceConfig extends RedisEnterpriseConfig {
     public static final String STREAM_CONSUMER_GROUP_DEFAULT = "kafka-consumer-group";
     public static final String STREAM_CONSUMER_GROUP_DOC = "Stream consumer group";
 
+    public static final String STREAM_CONSUMER_NAME = "redis.stream.consumer.name";
+    public static final String STREAM_CONSUMER_NAME_DEFAULT = "consumer-" + TOKEN_TASK;
+    public static final String STREAM_CONSUMER_NAME_DOC = "A format string for the stream consumer, which may contain '" + TOKEN_TASK + "' as a placeholder for the task id.\nFor example, 'consumer-" + TOKEN_TASK + "' for the task id '123' will map to the consumer name 'consumer-123'.";
+
+
     public static final String STREAM_BLOCK = "redis.stream.block";
     public static final long STREAM_BLOCK_DEFAULT = 100;
     public static final String STREAM_BLOCK_DOC = "The max amount of time in milliseconds to wait while polling for stream messages (XREAD [BLOCK milliseconds])";
@@ -66,6 +70,7 @@ public class RedisEnterpriseSourceConfig extends RedisEnterpriseConfig {
     private final String streamName;
     private final String streamOffset;
     private final String streamConsumerGroup;
+    private final String streamConsumerName;
     private final Long batchSize;
     private final Long streamBlock;
     private final String topicName;
@@ -79,16 +84,10 @@ public RedisEnterpriseSourceConfig(Map<?, ?> originals) {
         this.streamName = getString(STREAM_NAME);
         this.streamOffset = getString(STREAM_OFFSET);
         this.streamConsumerGroup = getString(STREAM_CONSUMER_GROUP);
+        this.streamConsumerName = getString(STREAM_CONSUMER_NAME);
         this.streamBlock = getLong(STREAM_BLOCK);
     }
 
-    private List<String> splitCommaSeparated(String string) {
-        if (string.trim().isEmpty()) {
-            return Collections.emptyList();
-        }
-        return Arrays.asList(string.split(","));
-    }
-
     public ReaderType getReaderType() {
         return readerType;
     }
@@ -117,6 +116,10 @@ public String getStreamConsumerGroup() {
         return streamConsumerGroup;
     }
 
+    public String getStreamConsumerName() {
+        return streamConsumerName;
+    }
+
     public String getTopicName() {
         return topicName;
     }
@@ -133,13 +136,14 @@ public RedisEnterpriseSourceConfigDef(ConfigDef base) {
         }
 
         private void define() {
-            define(ConfigKeyBuilder.of(TOPIC, ConfigDef.Type.STRING).importance(ConfigDef.Importance.MEDIUM).documentation(TOPIC_DOC).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(KEY_PATTERNS, Type.LIST).documentation(KEY_PATTERNS_DOC).defaultValue(KEY_PATTERNS_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(TOPIC, ConfigDef.Type.STRING).defaultValue(TOPIC_DEFAULT).importance(ConfigDef.Importance.MEDIUM).documentation(TOPIC_DOC).build());
+            define(ConfigKeyBuilder.of(BATCH_SIZE, ConfigDef.Type.LONG).defaultValue(BATCH_SIZE_DEFAULT).importance(ConfigDef.Importance.LOW).documentation(BATCH_SIZE_DOC).validator(ConfigDef.Range.atLeast(1L)).build());
+            define(ConfigKeyBuilder.of(READER_TYPE, ConfigDef.Type.STRING).documentation(READER_DOC).defaultValue(READER_DEFAULT).importance(ConfigDef.Importance.HIGH).validator(Validators.validEnum(ReaderType.class)).internalConfig(true).build());
+            define(ConfigKeyBuilder.of(KEY_PATTERNS, Type.LIST).documentation(KEY_PATTERNS_DOC).defaultValue(KEY_PATTERNS_DEFAULT).importance(Importance.MEDIUM).internalConfig(true).build());
+            define(ConfigKeyBuilder.of(STREAM_NAME, ConfigDef.Type.STRING).documentation(STREAM_NAME_DOC).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_CONSUMER_GROUP, ConfigDef.Type.STRING).documentation(STREAM_CONSUMER_GROUP_DOC).defaultValue(STREAM_CONSUMER_GROUP_DEFAULT).importance(ConfigDef.Importance.MEDIUM).build());
-            define(ConfigKeyBuilder.of(BATCH_SIZE, ConfigDef.Type.LONG).defaultValue(BATCH_SIZE_DEFAULT).importance(ConfigDef.Importance.LOW).documentation(BATCH_SIZE_DOC).validator(ConfigDef.Range.atLeast(1L)).build());
+            define(ConfigKeyBuilder.of(STREAM_CONSUMER_NAME, ConfigDef.Type.STRING).documentation(STREAM_CONSUMER_NAME_DOC).defaultValue(STREAM_CONSUMER_NAME_DEFAULT).importance(ConfigDef.Importance.MEDIUM).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());
         }
 

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

@@ -23,20 +23,20 @@ public class StreamSourceRecordReader extends AbstractSourceRecordReader<StreamM
     private static final String VALUE_SCHEMA_NAME = "com.redislabs.kafka.connect.StreamEventValue";
     private static final Schema VALUE_SCHEMA = SchemaBuilder.map(Schema.STRING_SCHEMA, Schema.STRING_SCHEMA).name(VALUE_SCHEMA_NAME);
     private final String topic;
-    private final int taskId;
+    private final String consumer;
 
     private StreamItemReader reader;
 
     public StreamSourceRecordReader(RedisEnterpriseSourceConfig sourceConfig, int taskId) {
         super(sourceConfig);
         this.topic = sourceConfig.getTopicName().replace(RedisEnterpriseSourceConfig.TOKEN_STREAM, sourceConfig.getStreamName());
-        this.taskId = taskId;
+        this.consumer = sourceConfig.getStreamConsumerName().replace(RedisEnterpriseSourceConfig.TOKEN_TASK, String.valueOf(taskId));
     }
 
     @Override
     protected void open(RedisClient client) {
         XReadArgs.StreamOffset<String> streamOffset = XReadArgs.StreamOffset.from(sourceConfig.getStreamName(), sourceConfig.getStreamOffset());
-        reader = StreamItemReader.client(client).offset(streamOffset).block(Duration.ofMillis(sourceConfig.getStreamBlock())).count(sourceConfig.getBatchSize()).consumerGroup(sourceConfig.getStreamConsumerGroup()).consumer("consumer" + taskId).build();
+        reader = StreamItemReader.client(client).offset(streamOffset).block(Duration.ofMillis(sourceConfig.getStreamBlock())).count(sourceConfig.getBatchSize()).consumerGroup(sourceConfig.getStreamConsumerGroup()).consumer(consumer).build();
         reader.open(new ExecutionContext());
     }
 

src/test/integration/java/com/redislabs/kafka/connect/RedisEnterpriseSourceConnectorIT.java

@@ -20,18 +20,21 @@ public void taskConfigsReturnsPartitionedTaskConfigs(RedisServer server) {
 
         final Map<String, String> connectorConfig = new HashMap<>();
         connectorConfig.put(RedisEnterpriseSourceConfig.READER_TYPE, RedisEnterpriseSourceConfig.ReaderType.KEYS.name());
+        connectorConfig.put(RedisEnterpriseSourceConfig.STREAM_NAME, "dummy");
         connectorConfig.put(RedisEnterpriseSourceConfig.TOPIC, "mytopic");
         connectorConfig.put(RedisEnterpriseSourceConfig.REDIS_URI, server.getRedisURI());
         connectorConfig.put(RedisEnterpriseSourceConfig.KEY_PATTERNS, "a:*,b:*,c:*");
 
         final Map<String, String> expectedPartitionedConnectorConfigA = new HashMap<>();
         expectedPartitionedConnectorConfigA.put(RedisEnterpriseSourceConfig.READER_TYPE, RedisEnterpriseSourceConfig.ReaderType.KEYS.name());
+        expectedPartitionedConnectorConfigA.put(RedisEnterpriseSourceConfig.STREAM_NAME, "dummy");
         expectedPartitionedConnectorConfigA.put(RedisEnterpriseSourceConfig.TOPIC, "mytopic");
         expectedPartitionedConnectorConfigA.put(RedisEnterpriseSourceConfig.REDIS_URI, server.getRedisURI());
         expectedPartitionedConnectorConfigA.put(RedisEnterpriseSourceConfig.KEY_PATTERNS, "a:*,b:*");
 
         final Map<String, String> expectedPartitionedConnectorConfigB = new HashMap<>();
         expectedPartitionedConnectorConfigB.put(RedisEnterpriseSourceConfig.READER_TYPE, RedisEnterpriseSourceConfig.ReaderType.KEYS.name());
+        expectedPartitionedConnectorConfigB.put(RedisEnterpriseSourceConfig.STREAM_NAME, "dummy");
         expectedPartitionedConnectorConfigB.put(RedisEnterpriseSourceConfig.TOPIC, "mytopic");
         expectedPartitionedConnectorConfigB.put(RedisEnterpriseSourceConfig.REDIS_URI, server.getRedisURI());
         expectedPartitionedConnectorConfigB.put(RedisEnterpriseSourceConfig.KEY_PATTERNS, "c:*");

src/test/integration/java/com/redislabs/kafka/connect/RedisEnterpriseSourceTaskIT.java

@@ -63,7 +63,7 @@ public void pollStream(RedisServer redis) throws InterruptedException {
     @MethodSource("redisServers")
     public void pollKeys(RedisServer redis) throws InterruptedException {
         String topic = "mytopic";
-        startTask(redis, RedisEnterpriseSourceConfig.TOPIC, topic, RedisEnterpriseSourceTask.KEYS_IDLE_TIMEOUT, "800");
+        startTask(redis, RedisEnterpriseSourceConfig.READER_TYPE, RedisEnterpriseSourceConfig.ReaderType.KEYS.name(), RedisEnterpriseSourceConfig.STREAM_NAME, "dummy", RedisEnterpriseSourceConfig.TOPIC, topic, RedisEnterpriseSourceTask.KEYS_IDLE_TIMEOUT, "800");
         Thread.sleep(500);
         String stringKey = "key:1";
         String stringValue = "my string";