Cluster Topology Refresh + General Maintenance (#19)

Author: jaredpetersen

CHANGELOG.md

@@ -4,6 +4,15 @@ 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.2.0] - 2021-02-13
+### Added
+- Handle Redis cluster topology changes on the fly
+
+### Changed
+- Upgraded various dependencies
+- Upgraded demo to use Confluent Platform 6.1.0
+- Integration testing improvements
+
 ## [1.1.0] - 2020-12-11
 ### Added
 - Parallelization for source connector based on channels/patterns

LICENSE.md

@@ -1,6 +1,6 @@
 The MIT License (MIT)
 
-Copyright (c) 2020 Jared Petersen
+Copyright (c) 2021 Jared Petersen
 
 Permission is hereby granted, free of charge, to any person obtaining a copy
 of this software and associated documentation files (the "Software"), to deal

docs/demo/README.md

@@ -27,7 +27,7 @@ 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://repo1.maven.org/maven2/io/github/jaredpetersen/kafka-connect-redis/1.1.0/kafka-connect-redis-1.1.0.jar
+curl -O https://repo1.maven.org/maven2/io/github/jaredpetersen/kafka-connect-redis/1.2.0/kafka-connect-redis-1.2.0.jar
 docker build -t jaredpetersen/kafka-connect-redis:latest .
 ```
 
@@ -50,7 +50,7 @@ Be patient, this can take a few minutes.
 
 Run the following command to configure Redis to run in cluster mode instead of standalone mode:
 ```bash
-kubectl -n kcr-demo run -it --rm redis-client --image redis:6 -- redis-cli --pass IEPfIr0eLF7UsfwrIlzy80yUaBG258j9 --cluster create $(kubectl -n kcr-demo get pods -l app=redis-cluster -o jsonpath='{range.items[*]}{.status.podIP}:6379 ') --cluster-yes
+kubectl -n kcr-demo run -it --rm redis-client --image redis:6 -- redis-cli --pass IEPfIr0eLF7UsfwrIlzy80yUaBG258j9 --cluster create $(kubectl -n kcr-demo get pods -l app=redis-cluster -o jsonpath='{range.items[*]}{.status.podIP}:6379 {end}') --cluster-yes
 ```
 
 ## Usage

docs/demo/SINK.md

@@ -1,24 +1,31 @@
 # Demo: Kafka Connect Sink
 ## Install Connector
-Send a request to the Kafka Connect REST API to configure it to use Kafka Connect Redis:
+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.
+First, expose the Kafka Connect server:
+```bash
+kubectl -n kcr-demo port-forward service/kafka-connect :rest
+```
+
+Kubectl will choose an available port for you that you will need to use for the cURLs (`$PORT`).
 
+### Avro
 ```bash
 curl --request POST \
-    --url "$(minikube -n kcr-demo service kafka-connect --url)/connectors" \
+    --url "localhost:$PORT/connectors" \
     --header 'content-type: application/json' \
     --data '{
         "name": "demo-redis-sink-connector",
         "config": {
             "connector.class": "io.github.jaredpetersen.kafkaconnectredis.sink.RedisSinkConnector",
             "key.converter": "io.confluent.connect.avro.AvroConverter",
             "key.converter.schema.registry.url": "http://kafka-schema-registry:8081",
+            "key.converter.key.subject.name.strategy": "io.confluent.kafka.serializers.subject.TopicRecordNameStrategy",
             "value.converter": "io.confluent.connect.avro.AvroConverter",
             "value.converter.schema.registry.url": "http://kafka-schema-registry:8081",
+            "value.converter.value.subject.name.strategy": "io.confluent.kafka.serializers.subject.TopicRecordNameStrategy",
             "tasks.max": "3",
-            "topics": "redis.commands.set,redis.commands.expire,redis.commands.expireat,redis.commands.pexpire,redis.commands.sadd,redis.commands.geoadd,redis.commands.arbitrary",
+            "topics": "redis.commands",
             "redis.uri": "redis://IEPfIr0eLF7UsfwrIlzy80yUaBG258j9@redis-cluster",
             "redis.cluster.enabled": true
         }
@@ -28,7 +35,7 @@ curl --request POST \
 ### Connect JSON
 ```bash
 curl --request POST \
-    --url "$(minikube -n kcr-demo service kafka-connect --url)/connectors" \
+    --url "localhost:$PORT/connectors" \
     --header 'content-type: application/json' \
     --data '{
         "name": "demo-redis-sink-connector",
@@ -48,17 +55,18 @@ curl --request POST \
 ### Avro
 Create an interactive ephemeral query pod:
 ```bash
-kubectl -n kcr-demo run -it --rm kafka-write-records --image confluentinc/cp-schema-registry:6.0.0 --command /bin/bash
+kubectl -n kcr-demo run -it --rm kafka-write-records --image confluentinc/cp-schema-registry:6.1.0 --command /bin/bash
 ```
 
-Write records to the `redis.commands` topics:
+Write records to the `redis.commands` topic:
 
 ```bash
 kafka-avro-console-producer \
     --broker-list kafka-broker-0.kafka-broker:9092 \
     --property schema.registry.url='http://kafka-schema-registry:8081' \
+    --property value.subject.name.strategy='io.confluent.kafka.serializers.subject.TopicRecordNameStrategy' \
     --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}]}' \
-    --topic redis.commands.set
+    --topic redis.commands
 >{"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}
@@ -70,35 +78,39 @@ kafka-avro-console-producer \
 kafka-avro-console-producer \
     --broker-list kafka-broker-0.kafka-broker:9092 \
     --property schema.registry.url='http://kafka-schema-registry:8081' \
+    --property value.subject.name.strategy='io.confluent.kafka.serializers.subject.TopicRecordNameStrategy' \
     --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
+    --topic redis.commands
 >{"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.subject.name.strategy='io.confluent.kafka.serializers.subject.TopicRecordNameStrategy' \
     --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
+    --topic redis.commands
 >{"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.subject.name.strategy='io.confluent.kafka.serializers.subject.TopicRecordNameStrategy' \
     --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
+    --topic redis.commands
 >{"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.subject.name.strategy='io.confluent.kafka.serializers.subject.TopicRecordNameStrategy' \
     --property value.schema='{"namespace":"io.github.jaredpetersen.kafkaconnectredis","name":"RedisSaddCommand","type":"record","fields":[{"name":"key","type":"string"},{"name":"values","type":{"type":"array","items":"string"}}]}' \
-    --topic redis.commands.sadd
+    --topic redis.commands
 >{"key":"{user.1}.interests","values":["reading"]}
 >{"key":"{user.2}.interests","values":["sailing","woodworking","programming"]}
 ```
@@ -107,17 +119,19 @@ kafka-avro-console-producer \
 kafka-avro-console-producer \
     --broker-list kafka-broker-0.kafka-broker:9092 \
     --property schema.registry.url='http://kafka-schema-registry:8081' \
+    --property value.subject.name.strategy='io.confluent.kafka.serializers.subject.TopicRecordNameStrategy' \
     --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"}]}}}]}' \
-    --topic redis.commands.geoadd
+    --topic redis.commands
 >{"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.subject.name.strategy='io.confluent.kafka.serializers.subject.TopicRecordNameStrategy' \
     --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
+    --topic redis.commands
 >{"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"]}
@@ -126,7 +140,7 @@ kafka-avro-console-producer \
 ### Connect JSON
 Create an interactive ephemeral query pod:
 ```bash
-kubectl -n kcr-demo run -it --rm kafka-write-records --image confluentinc/cp-kafka:6.0.0 --command /bin/bash
+kubectl -n kcr-demo run -it --rm kafka-write-records --image confluentinc/cp-kafka:6.1.0 --command /bin/bash
 ```
 
 Write records to the `redis.commands` topic:

docs/demo/SOURCE.md

@@ -2,10 +2,17 @@
 ## Install Connector
 Send a request to the Kafka Connect REST API to configure it to use Kafka Connect Redis. We'll be listening to all keyspace and keyevent notifications in Redis by using the channel pattern `__key*__:*`.
 
+First, expose the Kafka Connect server:
+```bash
+kubectl -n kcr-demo port-forward service/kafka-connect :rest
+```
+
+Kubectl will choose an available port for you that you will need to use for the cURLs (`$PORT`).
+
 ### Avro
 ```bash
 curl --request POST \
-    --url "$(minikube -n kcr-demo service kafka-connect --url)/connectors" \
+    --url "localhost:$PORT/connectors" \
     --header 'content-type: application/json' \
     --data '{
         "name": "demo-redis-source-connector",
@@ -28,7 +35,7 @@ curl --request POST \
 ### Connect JSON
 ```bash
 curl --request POST \
-    --url "$(minikube -n kcr-demo service kafka-connect --url)/connectors" \
+    --url "localhost:$PORT/connectors" \
     --header 'content-type: application/json' \
     --data '{
         "name": "demo-redis-source-connector",
@@ -75,7 +82,7 @@ SMEMBERS {user.2}.interests
 ### Avro
 Create an interactive ephemeral query pod:
 ```bash
-kubectl -n kcr-demo run -it --rm kafka-tail-records --image confluentinc/cp-schema-registry:6.0.0 --command /bin/bash
+kubectl -n kcr-demo run -it --rm kafka-tail-records --image confluentinc/cp-schema-registry:6.1.0 --command /bin/bash
 ```
 
 Tail the topic, starting from the beginning:
@@ -92,7 +99,7 @@ kafka-avro-console-consumer \
 ### Connect JSON
 Create an interactive ephemeral query pod:
 ```bash
-kubectl -n kcr-demo run -it --rm kafka-tail-records --image confluentinc/cp-kafka:6.0.0 --command /bin/bash
+kubectl -n kcr-demo run -it --rm kafka-tail-records --image confluentinc/cp-kafka:6.1.0 --command /bin/bash
 ```
 
 Tail the topic, starting from the beginning:

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

@@ -1,3 +1,3 @@
-ARG VERSION=6.0.0
+ARG VERSION=6.1.0
 FROM confluentinc/cp-kafka-connect-base:${VERSION}
 COPY kafka-connect-redis-*.jar /usr/share/java/kafka-connect-redis/

docs/demo/kubernetes/kafka-broker/statefulset.yaml

@@ -15,7 +15,7 @@ spec:
     spec:
       containers:
       - name: kafka-broker
-        image: confluentinc/cp-kafka:6.0.0
+        image: confluentinc/cp-kafka:6.1.0
         command:
         - /bin/bash
         args:

docs/demo/kubernetes/schema-registry/deployment.yaml

@@ -14,7 +14,7 @@ spec:
     spec:
       containers:
       - name: kafka-schema-registry
-        image: confluentinc/cp-schema-registry:6.0.0
+        image: confluentinc/cp-schema-registry:6.1.0
         ports:
         - containerPort: 8081
           name: rest

docs/demo/kubernetes/zookeeper/statefulset.yaml

@@ -15,7 +15,7 @@ spec:
     spec:
       containers:
       - name: zookeeper
-        image: confluentinc/cp-zookeeper:6.0.0
+        image: confluentinc/cp-zookeeper:6.1.0
         # Use downward api after https://github.com/kubernetes/kubernetes/pull/68719 is merged
         command:
         - /bin/bash

pom.xml

@@ -5,7 +5,7 @@
 
   <groupId>io.github.jaredpetersen</groupId>
   <artifactId>kafka-connect-redis</artifactId>
-  <version>1.1.0</version>
+  <version>1.2.0</version>
   <packaging>jar</packaging>
 
   <name>kafka-connect-redis</name>
@@ -62,26 +62,26 @@
     <dependency>
       <groupId>org.apache.kafka</groupId>
       <artifactId>connect-api</artifactId>
-      <version>2.6.0</version>
+      <version>2.7.0</version>
       <scope>provided</scope>
     </dependency>
 
     <dependency>
       <groupId>io.lettuce</groupId>
       <artifactId>lettuce-core</artifactId>
-      <version>6.0.1.RELEASE</version>
+      <version>6.0.2.RELEASE</version>
     </dependency>
 
     <dependency>
       <groupId>io.projectreactor</groupId>
       <artifactId>reactor-core</artifactId>
-      <version>3.3.10.RELEASE</version>
+      <version>3.4.2</version>
     </dependency>
 
     <dependency>
       <groupId>org.projectlombok</groupId>
       <artifactId>lombok</artifactId>
-      <version>1.18.16</version>
+      <version>1.18.18</version>
       <scope>provided</scope>
     </dependency>
 
@@ -95,34 +95,34 @@
     <dependency>
       <groupId>org.junit.jupiter</groupId>
       <artifactId>junit-jupiter</artifactId>
-      <version>5.7.0</version>
+      <version>5.7.1</version>
       <scope>test</scope>
     </dependency>
 
     <dependency>
       <groupId>org.mockito</groupId>
       <artifactId>mockito-core</artifactId>
-      <version>3.5.15</version>
+      <version>3.7.7</version>
       <scope>test</scope>
     </dependency>
 
     <dependency>
       <groupId>org.testcontainers</groupId>
       <artifactId>testcontainers</artifactId>
-      <version>1.15.0-rc2</version>
+      <version>1.15.2</version>
       <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>org.testcontainers</groupId>
       <artifactId>junit-jupiter</artifactId>
-      <version>1.15.0-rc2</version>
+      <version>1.15.2</version>
       <scope>test</scope>
     </dependency>
 
     <dependency>
       <groupId>io.projectreactor</groupId>
       <artifactId>reactor-test</artifactId>
-      <version>3.3.10.RELEASE</version>
+      <version>3.4.2</version>
       <scope>test</scope>
     </dependency>
   </dependencies>