Support sink expire commands (#12)

Author: jaredpetersen

CHANGELOG.md

@@ -4,6 +4,19 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [1.0.3] - 2020-11-21
+### Added
+- Added support for Redis EXPIRE commands
+- Added support for Redis EXPIREAT commands
+- Added support for Redis PEXPIRE commands
+
+### Changed
+- Improved source connector partitioning documentation
+
+### Fixed
+- Source connector no longer logs every Redis message at an INFO level
+- Added missing configuration property `topic` to the source connector documentation
+
 ## [1.0.2] - 2020-11-13
 ### Fixed
 - Fixed POM description to include source capability

README.md

@@ -11,7 +11,7 @@ Kafka Connect Redis Source subscribes to Redis channels/patterns (including [key
 For more information, see the [detailed documentation](/docs/connectors/SOURCE.md).
 
 ### Sink
-Kafka Connect Redis Sink consumes Kafka records in a Redis command format and applies them to Redis. Several write-based commands are supported at this time (`SET`, `SADD`, and `GEOADD`) and more are slated for the near future.
+Kafka Connect Redis Sink consumes Kafka records in a Redis command format (`SET`, `GEOADD`, etc.) and applies them to Redis.
 
 For more information, see the [detailed documentation](/docs/connectors/SINK.md).
 

docs/connectors/SINK.md

@@ -3,6 +3,9 @@ Consume messages from Kafka and apply them to Redis in the form of commands.
 
 The following commands are supported at this time:
 - [SET](https://redis.io/commands/set)
+- [EXPIRE](https://redis.io/commands/expire)
+- [EXPIREAT](https://redis.io/commands/expireat)
+- [PEXPIRE](https://redis.io/commands/pexpire)
 - [SADD](https://redis.io/commands/sadd)
 - [GEOADD](https://redis.io/commands/geoadd)
 
@@ -127,6 +130,126 @@ Keys are ignored.
 }
 ```
 
+#### EXPIRE
+##### Avro
+```json
+{
+    "namespace": "io.github.jaredpetersen.kafkaconnectredis",
+    "name": "RedisExpireCommand",
+    "type": "record",
+    "fields": [
+        {
+            "name": "key",
+            "type": "string"
+        },
+        {
+            "name": "seconds",
+            "type": "long"
+        }
+    ]
+}
+```
+
+##### Connect JSON
+```json
+{
+    "name": "io.github.jaredpetersen.kafkaconnectredis.RedisExpireCommand",
+    "type": "struct",
+    "fields": [
+        {
+            "field": "key",
+            "type": "string",
+            "optional": false
+        },
+        {
+            "field": "seconds",
+            "type": "int64",
+            "optional": false
+        }
+    ]
+}
+```
+
+#### EXPIREAT
+##### Avro
+```json
+{
+    "namespace": "io.github.jaredpetersen.kafkaconnectredis",
+    "name": "RedisExpireatCommand",
+    "type": "record",
+    "fields": [
+        {
+            "name": "key",
+            "type": "string"
+        },
+        {
+            "name": "timestamp",
+            "type": "long"
+        }
+    ]
+}
+```
+
+##### Connect JSON
+```json
+{
+    "name": "io.github.jaredpetersen.kafkaconnectredis.RedisExpireatCommand",
+    "type": "struct",
+    "fields": [
+        {
+            "field": "key",
+            "type": "string",
+            "optional": false
+        },
+        {
+            "field": "timestamp",
+            "type": "int64",
+            "optional": false
+        }
+    ]
+}
+```
+
+#### PEXPIRE
+##### Avro
+```json
+{
+    "namespace": "io.github.jaredpetersen.kafkaconnectredis",
+    "name": "RedisPexpireCommand",
+    "type": "record",
+    "fields": [
+        {
+            "name": "key",
+            "type": "string"
+        },
+        {
+            "name": "milliseconds",
+            "type": "long"
+        }
+    ]
+}
+```
+
+##### Connect JSON
+```json
+{
+    "name": "io.github.jaredpetersen.kafkaconnectredis.RedisPexpireCommand",
+    "type": "struct",
+    "fields": [
+        {
+            "field": "key",
+            "type": "string",
+            "optional": false
+        },
+        {
+            "field": "milliseconds",
+            "type": "int64",
+            "optional": false
+        }
+    ]
+}
+```
+
 #### SADD
 ##### Avro
 ```json

docs/connectors/SOURCE.md

@@ -77,16 +77,19 @@ Subscribes to Redis channels/patterns (including [keyspace notifications](https:
 ## Partitions
 Records are partitioned using the [`DefaultPartitioner`](https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/producer/internals/DefaultPartitioner.java) class. This means that the record key is used to determine which partition the record is assigned to.
 
-If you would prefer a different partitioning strategy, you may implement your own [`Partitioner`](https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/producer/Partitioner.java) and configure the connector to use it via [`partitioner.class`](https://kafka.apache.org/documentation/#partitioner.class). Alternatively, you may also implement a custom [Single Message Transform (SMT)](https://docs.confluent.io/current/connect/transforms/index.html).
+In the case of subscribing to Redis keyspace notifications, it may be useful to avoid partitioning the data so that multiple event types can arrive in order as a single event stream. This can be accomplished by configuring the connector to publish to a Kafka topic that only contains a single partition, forcing the DefaultPartitioner to only utilize the single partition.
+
+The plugin can be configured to use an alternative partitioning strategy if desired. Set the configuration property `connector.client.config.override.policy` to value `All` on the Kafka Connect worker (the overall Kafka Connect application that runs plugins). This will allow the override of the internal Kafka producer and consumer configurations. To override the partitioner for an individual connector plugin, add the configuration property `producer.override.partitioner.class` to the connector plugin with a value that points to a class implementing the [Partitioner](https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/producer/Partitioner.java) interface, e.g. `org.apache.kafka.clients.producer.internals.DefaultPartitioner`.
 
 ## Parallelization
 Splitting the workload between multiple tasks via the configuration property `max.tasks` is not supported at this time. Support for this will be added in the future.
 
 ## Configuration
 ### Connector Properties
-| Name                              | Type    | Default | Importance | Description                                             |
-| --------------------------------- | ------- | ------- | ---------- | ------------------------------------------------------- |
-| `redis.uri`                       | string  |         | High       | Redis connection information provided via a URI string. |
-| `redis.cluster.enabled`           | boolean | false   | High       | Target Redis is running as a cluster.                   |
-| `redis.channels`                  | string  |         | High       | Redis channels to subscribe to separated by commas.     |
-| `redis.channels.patterns.enabled` | boolean |         | High       | Redis channels use patterns (PSUBSCRIBE).               |
+| Name                              | Type    | Default        | Importance | Description                                             |
+| --------------------------------- | ------- | -------------- | ---------- | ------------------------------------------------------- |
+| `topic`                           | string  | `redis.events` | High       | Topic to write to.                                      |
+| `redis.uri`                       | string  |                | High       | Redis connection information provided via a URI string. |
+| `redis.cluster.enabled`           | boolean | false          | High       | Target Redis is running as a cluster.                   |
+| `redis.channels`                  | string  |                | High       | Redis channels to subscribe to separated by commas.     |
+| `redis.channels.patterns.enabled` | boolean |                | High       | Redis channels use patterns (PSUBSCRIBE).               |

docs/demo/README.md

@@ -15,7 +15,7 @@ Now that we have Kubernetes set up locally, we'll need a Docker image that conta
 
 Navigate to `demo/docker/` in this repository and run the following commands **in a separate terminal** to download the plugin and build the image for minikube:
 ```bash
-curl -O https://oss.sonatype.org/service/local/repositories/releases/content/io/github/jaredpetersen/kafka-connect-redis/1.0.2/kafka-connect-redis-1.0.1.jar
+curl -O https://oss.sonatype.org/service/local/repositories/releases/content/io/github/jaredpetersen/kafka-connect-redis/1.0.3/kafka-connect-redis-1.0.3.jar
 eval $(minikube docker-env)
 docker build -t jaredpetersen/kafka-connect-redis:latest .
 ```

docs/demo/SINK.md

@@ -3,6 +3,8 @@
 Send a request to the Kafka Connect REST API to configure it to use Kafka Connect Redis:
 
 ### Avro
+**IMPORTANT:** The Avro demo utilizes multiple topics in order to work around [a bug in the Avro console producer](https://github.com/confluentinc/schema-registry/issues/898). A fix has been merged but Confluent has not published a new Docker image for it yet (6.1.0+). Kafka Connect Redis works with Avro on a single topic; this is just a problem with the console producer provided by Confluent.
+
 ```bash
 curl --request POST \
     --url "$(minikube -n kcr-demo service kafka-connect --url)/connectors" \
@@ -16,7 +18,7 @@ curl --request POST \
             "value.converter": "io.confluent.connect.avro.AvroConverter",
             "value.converter.schema.registry.url": "http://kafka-schema-registry:8081",
             "tasks.max": "1",
-            "topics": "redis.commands",
+            "topics": "redis.commands.set,redis.commands.expire,redis.commands.expireat,redis.commands.pexpire,redis.commands.sadd,redis.commands.geoadd",
             "redis.uri": "redis://IEPfIr0eLF7UsfwrIlzy80yUaBG258j9@redis-cluster",
             "redis.cluster.enabled": true
         }
@@ -49,28 +51,54 @@ Create an interactive ephemeral query pod:
 kubectl -n kcr-demo run -it --rm kafka-write-records --image confluentinc/cp-schema-registry:6.0.0 --command /bin/bash
 ```
 
-Write records to the `redis.commands` topic:
-
-**IMPORTANT:** The following Avro example does not completely work [due to a bug in the Avro console producer](https://github.com/confluentinc/schema-registry/issues/898). A fix has been merged but Confluent has not published a new Docker image for it yet (6.1.0+). In the meantime, you may only issue one Redis command type per topic if you are using the Avro console producer.
+Write records to the `redis.commands` topics:
 
 ```bash
 kafka-avro-console-producer \
     --broker-list kafka-broker-0.kafka-broker:9092 \
     --property schema.registry.url='http://kafka-schema-registry:8081' \
     --property value.schema='{"namespace":"io.github.jaredpetersen.kafkaconnectredis","name":"RedisSetCommand","type":"record","fields":[{"name":"key","type":"string"},{"name":"value","type":"string"},{"name":"expiration","type":["null",{"name":"RedisSetCommandExpiration","type":"record","fields":[{"name":"type","type":{"name":"RedisSetCommandExpirationType","type":"enum","symbols":["EX","PX","KEEPTTL"]}},{"name":"time","type":["null","long"]}]}],"default":null},{"name":"condition","type":["null",{"name":"RedisSetCommandCondition","type":"enum","symbols":["NX","XX","KEEPTTL"]}],"default":null}]}' \
-    --property value.subject.name.strategy='io.confluent.kafka.serializers.subject.TopicRecordNameStrategy' \
-    --topic redis.commands
+    --topic redis.commands.set
 >{"key":"{user.1}.username","value":"jetpackmelon22","expiration":null,"condition":null}
 >{"key":"{user.2}.username","value":"anchorgoat74","expiration":{"io.github.jaredpetersen.kafkaconnectredis.RedisSetCommandExpiration":{"type":"EX","time":{"long":2100}}},"condition":{"io.github.jaredpetersen.kafkaconnectredis.RedisSetCommandCondition":"NX"}}
+>{"key":"product.milk","value":"$2.29","expiration":null,"condition":null}
+>{"key":"product.bread","value":"$5.49","expiration":null,"condition":null}
+>{"key":"product.waffles","value":"$2.59","expiration":null,"condition":null}
+```
+
+```bash
+kafka-avro-console-producer \
+    --broker-list kafka-broker-0.kafka-broker:9092 \
+    --property schema.registry.url='http://kafka-schema-registry:8081' \
+    --property value.schema='{"namespace":"io.github.jaredpetersen.kafkaconnectredis","name":"RedisExpireCommand","type":"record","fields":[{"name":"key","type":"string"},{"name":"seconds","type":"long"}]}' \
+    --topic redis.commands.expire
+>{"key":"product.milk","seconds":1800}
+```
+
+```bash
+kafka-avro-console-producer \
+    --broker-list kafka-broker-0.kafka-broker:9092 \
+    --property schema.registry.url='http://kafka-schema-registry:8081' \
+    --property value.schema='{"namespace":"io.github.jaredpetersen.kafkaconnectredis","name":"RedisExpireatCommand","type":"record","fields":[{"name":"key","type":"string"},{"name":"timestamp","type":"long"}]}' \
+    --topic redis.commands.expireat
+>{"key":"product.bread","timestamp":4130464553}
+```
+
+```bash
+kafka-avro-console-producer \
+    --broker-list kafka-broker-0.kafka-broker:9092 \
+    --property schema.registry.url='http://kafka-schema-registry:8081' \
+    --property value.schema='{"namespace":"io.github.jaredpetersen.kafkaconnectredis","name":"RedisPexpireCommand","type":"record","fields":[{"name":"key","type":"string"},{"name":"milliseconds","type":"long"}]}' \
+    --topic redis.commands.pexpire
+>{"key":"product.waffles","milliseconds":1800000}
 ```
 
 ```bash
 kafka-avro-console-producer \
     --broker-list kafka-broker-0.kafka-broker:9092 \
     --property schema.registry.url='http://kafka-schema-registry:8081' \
     --property value.schema='{"namespace":"io.github.jaredpetersen.kafkaconnectredis","name":"RedisSaddCommand","type":"record","fields":[{"name":"key","type":"string"},{"name":"values","type":{"type":"array","items":"string"}}]}' \
-    --property value.subject.name.strategy='io.confluent.kafka.serializers.subject.TopicRecordNameStrategy' \
-    --topic redis.commands
+    --topic redis.commands.sadd
 >{"key":"{user.1}.interests","values":["reading"]}
 >{"key":"{user.2}.interests","values":["sailing","woodworking","programming"]}
 ```
@@ -80,8 +108,7 @@ kafka-avro-console-producer \
     --broker-list kafka-broker-0.kafka-broker:9092 \
     --property schema.registry.url='http://kafka-schema-registry:8081' \
     --property value.schema='{"namespace":"io.github.jaredpetersen.kafkaconnectredis","name":"RedisGeoaddCommand","type":"record","fields":[{"name":"key","type":"string"},{"name":"values","type":{"type":"array","items":{"name":"RedisGeoaddCommandGeolocation","type":"record","fields":[{"name":"longitude","type":"double"},{"name":"latitude","type":"double"},{"name":"member","type":"string"}]}}}]}' \
-    --property value.subject.name.strategy='io.confluent.kafka.serializers.subject.TopicRecordNameStrategy' \
-    --topic redis.commands
+    --topic redis.commands.geoadd
 >{"key":"Sicily","values":[{"longitude":13.361389,"latitude":13.361389,"member":"Palermo"},{"longitude":15.087269,"latitude":37.502669,"member":"Catania"}]}
 ```
 
@@ -98,6 +125,12 @@ kafka-console-producer \
     --topic redis.commands
 >{"payload":{"key":"{user.1}.username","value":"jetpackmelon22"},"schema":{"name":"io.github.jaredpetersen.kafkaconnectredis.RedisSetCommand","type":"struct","fields":[{"field":"key","type":"string","optional":false},{"field":"value","type":"string","optional":false},{"field":"expiration","type":"struct","fields":[{"field":"type","type":"string","optional":false},{"field":"time","type":"int64","optional":true}],"optional":true},{"field":"condition","type":"string","optional":true}]}}
 >{"payload":{"key":"{user.2}.username","value":"anchorgoat74","expiration":{"type":"EX","time":2100},"condition":"NX"},"schema":{"name":"io.github.jaredpetersen.kafkaconnectredis.RedisSetCommand","type":"struct","fields":[{"field":"key","type":"string","optional":false},{"field":"value","type":"string","optional":false},{"field":"expiration","type":"struct","fields":[{"field":"type","type":"string","optional":false},{"field":"time","type":"int64","optional":true}],"optional":true},{"field":"condition","type":"string","optional":true}]}}
+>{"payload":{"key":"product.milk","value":"$2.29"},"schema":{"name":"io.github.jaredpetersen.kafkaconnectredis.RedisSetCommand","type":"struct","fields":[{"field":"key","type":"string","optional":false},{"field":"value","type":"string","optional":false},{"field":"expiration","type":"struct","fields":[{"field":"type","type":"string","optional":false},{"field":"time","type":"int64","optional":true}],"optional":true},{"field":"condition","type":"string","optional":true}]}}
+>{"payload":{"key":"product.milk","seconds":1800},"schema":{"name":"io.github.jaredpetersen.kafkaconnectredis.RedisExpireCommand","type":"struct","fields":[{"field":"key","type":"string","optional":false},{"field":"seconds","type":"int64","optional":false}]}}
+>{"payload":{"key":"product.bread","value":"$5.49"},"schema":{"name":"io.github.jaredpetersen.kafkaconnectredis.RedisSetCommand","type":"struct","fields":[{"field":"key","type":"string","optional":false},{"field":"value","type":"string","optional":false},{"field":"expiration","type":"struct","fields":[{"field":"type","type":"string","optional":false},{"field":"time","type":"int64","optional":true}],"optional":true},{"field":"condition","type":"string","optional":true}]}}
+>{"payload":{"key":"product.bread","timestamp":4130464553},"schema":{"name":"io.github.jaredpetersen.kafkaconnectredis.RedisExpireatCommand","type":"struct","fields":[{"field":"key","type":"string","optional":false},{"field":"timestamp","type":"int64","optional":false}]}}
+>{"payload":{"key":"product.waffles","value":"$2.59"},"schema":{"name":"io.github.jaredpetersen.kafkaconnectredis.RedisSetCommand","type":"struct","fields":[{"field":"key","type":"string","optional":false},{"field":"value","type":"string","optional":false},{"field":"expiration","type":"struct","fields":[{"field":"type","type":"string","optional":false},{"field":"time","type":"int64","optional":true}],"optional":true},{"field":"condition","type":"string","optional":true}]}}
+>{"payload":{"key":"product.waffles","milliseconds":1800000},"schema":{"name":"io.github.jaredpetersen.kafkaconnectredis.RedisPexpireCommand","type":"struct","fields":[{"field":"key","type":"string","optional":false},{"field":"milliseconds","type":"int64","optional":false}]}}
 >{"payload":{"key":"{user.1}.interests","values":["reading"]},"schema":{"name":"io.github.jaredpetersen.kafkaconnectredis.RedisSaddCommand","type":"struct","fields":[{"field":"key","type":"string","optional":false},{"field":"values","type":"array","items":{"type":"string"},"optional":false}]}}
 >{"payload":{"key":"{user.2}.interests","values":["sailing","woodworking","programming"]},"schema":{"name":"io.github.jaredpetersen.kafkaconnectredis.RedisSaddCommand","type":"struct","fields":[{"field":"key","type":"string","optional":false},{"field":"values","type":"array","items":{"type":"string"},"optional":false}]}}
 >{"payload":{"key":"Sicily","values":[{"longitude":13.361389,"latitude":13.361389,"member":"Palermo"},{"longitude":15.087269,"latitude":37.502669,"member":"Catania"}]},"schema":{"name":"io.github.jaredpetersen.kafkaconnectredis.RedisGeoaddCommand","type":"struct","fields":[{"field":"key","type":"string","optional":false},{"field":"values","type":"array","items":{"type":"struct","fields":[{"field":"longitude","type":"double","optional":false},{"field":"latitude","type":"double","optional":false},{"field":"member","type":"string","optional":false}]},"optional":false}]}}
@@ -118,6 +151,12 @@ Run commands to confirm the commands were applied correctly:
 ```bash
 GET {user.1}.username
 GET {user.2}.username
+GET product.bread
+TTL product.bread
+GET product.milk
+TTL product.milk
+GET product.waffles
+PTTL product.waffles
 SMEMBERS {user.1}.interests
 SMEMBERS {user.2}.interests
 GEOPOS Sicily Catania

pom.xml

@@ -5,7 +5,7 @@
 
   <groupId>io.github.jaredpetersen</groupId>
   <artifactId>kafka-connect-redis</artifactId>
-  <version>1.0.2</version>
+  <version>1.0.3</version>
   <packaging>jar</packaging>
 
   <name>kafka-connect-redis</name>