Support sinking arbitrary Redis commands (#16)

Author: jaredpetersen

CHANGELOG.md

@@ -4,6 +4,13 @@ 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.4] - 2020-11-29
+### Added
+- Added support for sinking arbitrary Redis commands, primarily for use with Redis modules
+
+### Fixed
+- Fixed linter configuration
+
 ## [1.0.3] - 2020-11-21
 ### Added
 - Added support for Redis EXPIRE commands

checkstyle.xml

@@ -61,6 +61,7 @@
             <property name="allowNonPrintableEscapes" value="true"/>
         </module>
         <module name="AvoidStarImport"/>
+        <module name="UnusedImports"/>
         <module name="OneTopLevelClass"/>
         <module name="NoLineWrap">
             <property name="tokens" value="PACKAGE_DEF, IMPORT, STATIC_IMPORT"/>

docs/connectors/SINK.md

@@ -8,6 +8,7 @@ The following commands are supported at this time:
 - [PEXPIRE](https://redis.io/commands/pexpire)
 - [SADD](https://redis.io/commands/sadd)
 - [GEOADD](https://redis.io/commands/geoadd)
+- Arbitrary -- useful for Redis modules
 
 Support for additional write-based commands will be added in the future.
 
@@ -376,6 +377,52 @@ Keys are ignored.
 }
 ```
 
+#### Arbitrary
+##### Avro
+```json
+{
+    "namespace": "io.github.jaredpetersen.kafkaconnectredis",
+    "name": "RedisArbitraryCommand",
+    "type": "record",
+    "fields": [
+        {
+            "name": "command",
+            "type": "string"
+        },
+        {
+            "name": "arguments",
+            "type": {
+                "type": "array",
+                "items": "string"
+            }
+        }
+    ]
+}
+```
+
+##### Connect JSON
+```json
+{
+    "name": "io.github.jaredpetersen.kafkaconnectredis.RedisArbitraryCommand",
+    "type": "struct",
+    "fields": [
+        {
+            "field": "command",
+            "type": "string",
+            "optional": false
+        },
+        {
+            "field": "arguments",
+            "type": "array",
+            "items": {
+              "type": "string"
+            },
+            "optional": false
+        }
+    ]
+}
+```
+
 ## Configuration
 ### Connector Properties
 | Name                    | Type    | Default | Importance | Description                                             |

docs/demo/README.md

@@ -11,12 +11,23 @@ minikube start --cpus 2 --memory 10g
 ```
 
 ### Docker
-Now that we have Kubernetes set up locally, we'll need a Docker image that contains Kafka Connect Redis.
+Now that we have Kubernetes set up locally, we'll need some Docker images.
 
-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:
+Open a new terminal we can use to build images for minikube. Run the following command to connect the terminal to minikube:
 ```bash
-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)
+```
+
+We'll use this terminal for the rest of this section.
+
+Let's start by building Redis. Navigate to `demo/docker/redis` and run the following commands:
+```bash
+docker build -t jaredpetersen/redis:latest .
+```
+
+Next, we'll need to build a docker image for Kafka Connect Redis. Navigate to `demo/docker/kafka-connect-redis` and run the following commands:
+```bash
+curl -O https://oss.sonatype.org/service/local/repositories/releases/content/io/github/jaredpetersen/kafka-connect-redis/1.0.4/kafka-connect-redis-1.0.4.jar
 docker build -t jaredpetersen/kafka-connect-redis:latest .
 ```
 

docs/demo/SINK.md

@@ -18,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.set,redis.commands.expire,redis.commands.expireat,redis.commands.pexpire,redis.commands.sadd,redis.commands.geoadd",
+            "topics": "redis.commands.set,redis.commands.expire,redis.commands.expireat,redis.commands.pexpire,redis.commands.sadd,redis.commands.geoadd,redis.commands.arbitrary",
             "redis.uri": "redis://IEPfIr0eLF7UsfwrIlzy80yUaBG258j9@redis-cluster",
             "redis.cluster.enabled": true
         }
@@ -112,6 +112,17 @@ kafka-avro-console-producer \
 >{"key":"Sicily","values":[{"longitude":13.361389,"latitude":13.361389,"member":"Palermo"},{"longitude":15.087269,"latitude":37.502669,"member":"Catania"}]}
 ```
 
+```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":"RedisArbitraryCommand","type":"record","fields":[{"name":"command","type":"string"},{"name":"arguments","type":{"type":"array","items":"string"}}]}' \
+    --topic redis.commands.arbitrary
+>{"command":"TS.CREATE","arguments":["temperature:3:11", "RETENTION", "60", "LABELS", "sensor_id", "2", "area_id", "32"]}
+>{"command":"TS.ADD","arguments":["temperature:3:11", "1548149181", "30"]}
+>{"command":"TS.ADD","arguments":["temperature:3:11", "1548149191", "42"]}
+```
+
 ### Connect JSON
 Create an interactive ephemeral query pod:
 ```bash
@@ -134,6 +145,9 @@ kafka-console-producer \
 >{"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}]}}
+>{"payload":{"command":"TS.CREATE","arguments":["temperature:3:11", "RETENTION", "60", "LABELS", "sensor_id", "2", "area_id", "32"]},"schema":{"name":"io.github.jaredpetersen.kafkaconnectredis.RedisArbitraryCommand","type":"struct","fields":[{"field":"command","type":"string","optional":false},{"field":"arguments","type":"array","items":{"type":"string"},"optional":false}]}}
+>{"payload":{"command":"TS.ADD","arguments":["temperature:3:11", "1548149181", "30"]},"schema":{"name":"io.github.jaredpetersen.kafkaconnectredis.RedisArbitraryCommand","type":"struct","fields":[{"field":"command","type":"string","optional":false},{"field":"arguments","type":"array","items":{"type":"string"},"optional":false}]}}
+>{"payload":{"command":"TS.ADD","arguments":["temperature:3:11", "1548149191", "42"]},"schema":{"name":"io.github.jaredpetersen.kafkaconnectredis.RedisArbitraryCommand","type":"struct","fields":[{"field":"command","type":"string","optional":false},{"field":"arguments","type":"array","items":{"type":"string"},"optional":false}]}}
 ```
 
 ## Validate
@@ -160,4 +174,5 @@ PTTL product.waffles
 SMEMBERS {user.1}.interests
 SMEMBERS {user.2}.interests
 GEOPOS Sicily Catania
+TS.RANGE temperature:3:11 1548149180 1548149210 AGGREGATION avg 5
 ```

docs/demo/docker/Dockerfile renamed to docs/demo/docker/kafka-connect-redis/Dockerfile

@@ -0,0 +1,9 @@
+FROM redislabs/redistimeseries:1.4.6 as redistimeseries
+FROM redis:6
+
+ENV LIBRARY_PATH /usr/lib/redis/modules
+
+COPY --from=redistimeseries ${LIBRARY_PATH}/redistimeseries.so ${LIBRARY_PATH}/redistimeseries.so
+
+ENTRYPOINT ["redis-server"]
+CMD ["/usr/local/etc/redis/redis.conf", "--loadmodule", "/usr/lib/redis/modules/redistimeseries.so"]

docs/demo/docker/redis/Dockerfile

@@ -16,9 +16,8 @@ spec:
     spec:
       containers:
       - name: redis
-        image: redis:6
-        args:
-        - /usr/local/etc/redis/redis.conf
+        image: jaredpetersen/redis:latest
+        imagePullPolicy: Never
         ports:
         - containerPort: 6379
           name: client

docs/demo/kubernetes/redis/statefulset.yaml

@@ -5,7 +5,7 @@
 
   <groupId>io.github.jaredpetersen</groupId>
   <artifactId>kafka-connect-redis</artifactId>
-  <version>1.0.3</version>
+  <version>1.0.4</version>
   <packaging>jar</packaging>
 
   <name>kafka-connect-redis</name>
@@ -69,7 +69,7 @@
     <dependency>
       <groupId>io.lettuce</groupId>
       <artifactId>lettuce-core</artifactId>
-      <version>6.0.0.RELEASE</version>
+      <version>6.0.1.RELEASE</version>
     </dependency>
 
     <dependency>
@@ -164,7 +164,8 @@
         <configuration>
           <configLocation>checkstyle.xml</configLocation>
           <consoleOutput>true</consoleOutput>
-          <failsOnError>true</failsOnError>
+          <failOnViolation>true</failOnViolation>
+          <violationSeverity>warning</violationSeverity>
           <includeTestSourceDirectory>true</includeTestSourceDirectory>
           <suppressionsLocation>checkstyle-test-suppressions.xml</suppressionsLocation>
         </configuration>
@@ -179,13 +180,6 @@
           <execution>
             <id>validate</id>
             <phase>validate</phase>
-            <configuration>
-              <configLocation>checkstyle.xml</configLocation>
-              <consoleOutput>true</consoleOutput>
-              <failsOnError>true</failsOnError>
-              <violationSeverity>warning</violationSeverity>
-              <includeTestSourceDirectory>true</includeTestSourceDirectory>
-            </configuration>
             <goals>
               <goal>check</goal>
             </goals>

pom.xml

@@ -0,0 +1,35 @@
+package io.github.jaredpetersen.kafkaconnectredis.sink.writer;
+
+import io.lettuce.core.codec.RedisCodec;
+import io.lettuce.core.output.CommandOutput;
+import java.nio.ByteBuffer;
+
+/**
+ * Void output of a Redis command used to "fire and forget".
+ * <p />
+ * Temporary stopgap until https://github.com/lettuce-io/lettuce-core/issues/1529 is released.
+ *
+ * @param <K> Key type.
+ * @param <V> Value type.
+ */
+class LettuceVoidOutput<K, V> extends CommandOutput<K, V, Void> {
+  public LettuceVoidOutput(RedisCodec<K, V> codec) {
+    super(codec, null);
+  }
+
+  @Override
+  public void set(ByteBuffer bytes) {
+  }
+
+  @Override
+  public void set(long integer) {
+  }
+
+  @Override
+  public void set(double number) {
+  }
+
+  @Override
+  public void set(boolean value) {
+  }
+}