feat: Added support for Redis cluster mode

Author: Julien Ruaux

.github/workflows/early-access.yml

@@ -7,7 +7,7 @@ on:
 jobs:
   earlyaccess:
     name: EarlyAccess
-    if: github.repository == 'redis-field-engineering/redis-enterprise-kafka' && startsWith(github.event.head_commit.message, 'Releasing version') != true
+    if: startsWith(github.event.head_commit.message, 'Releasing version') != true
     runs-on: ubuntu-18.04
     steps:
       - name: Checkout

README.adoc

@@ -1,4 +1,4 @@
-= Redis Enterprise Kafka Connector
+= Redis Kafka Connector
 :linkattrs:
 :project-owner:   redis-field-engineering
 :project-name:    redis-enterprise-kafka
@@ -8,7 +8,7 @@
 image:https://github.com/{project-owner}/{project-name}/actions/workflows/early-access.yml/badge.svg["Build Status", link="https://github.com/{project-owner}/{project-name}/actions"]
 image:https://codecov.io/gh/{project-owner}/{project-name}/branch/master/graph/badge.svg?token=MTMRRGEWBD["Coverage", link="https://codecov.io/gh/{project-owner}/{project-name}"]
 
-Kafka Connect source and sink connectors for https://redis.com/redis-enterprise-software/overview/[Redis Enterprise]
+Kafka Connect source and sink connectors for Redis
 
 == Documentation
 

jreleaser.yml

@@ -1,7 +1,7 @@
 project:
   name: redis-enterprise-kafka
-  description: Kafka Connector for Redis Enterprise
-  longDescription: Kafka Connect source and sink connectors for Redis Enterprise
+  description: Kafka Connector for Redis
+  longDescription: Kafka Connect source and sink connectors for Redis
   website: https://github.com/redis-field-engineering/redis-enterprise-kafka
   authors:
     - Julien Ruaux

pom.xml

@@ -7,8 +7,8 @@
 	<groupId>com.redis</groupId>
 	<artifactId>redis-enterprise-kafka</artifactId>
 	<version>6.8-SNAPSHOT</version>
-	<name>Redis Enterprise Kafka Connector</name>
-	<description>Kafka Connect source and sink connectors for Redis Enterprise</description>
+	<name>Redis Kafka Connector</name>
+	<description>Kafka Connect source and sink connectors for Redis</description>
 	<url>https://github.com/${github.owner}/${github.repo}</url>
 	<inceptionYear>2021</inceptionYear>
 	<properties>
@@ -83,12 +83,12 @@
 			<version>${kafka.version}</version>
 			<scope>provided</scope>
 		</dependency>
-	    <dependency>
-            <groupId>org.apache.kafka</groupId>
-            <artifactId>connect-json</artifactId>
-            <version>${kafka.version}</version>
-            <scope>provided</scope>
-        </dependency>
+		<dependency>
+			<groupId>org.apache.kafka</groupId>
+			<artifactId>connect-json</artifactId>
+			<version>${kafka.version}</version>
+			<scope>provided</scope>
+		</dependency>
 		<dependency>
 			<groupId>com.redis</groupId>
 			<artifactId>lettucemod</artifactId>
@@ -273,7 +273,7 @@
 				<artifactId>kafka-connect-maven-plugin</artifactId>
 				<version>0.12.0</version>
 				<configuration>
-					<title>Redis Enterprise Connector</title>
+					<title>Redis Connector (Source and Sink) by Redis</title>
 					<ownerUsername>redis</ownerUsername>
 					<documentationUrl>https://${github.owner}.github.io/${github.repo}/</documentationUrl>
 					<ownerLogo>src/docs/asciidoc/images/redis_logo.png</ownerLogo>
@@ -282,18 +282,16 @@
 					<sourceUrl>${project.scm.url}</sourceUrl>
 					<supportProviderName>Redis</supportProviderName>
 					<supportSummary>
-                        <![CDATA[This connector is <a href="https://redis.com/company/support/">supported by Redis</a> as part of a
-                                <a href="https://redis.com/redis-enterprise">Redis Enterprise</a> license.]]>
+                        <![CDATA[Contact us on the <a href="https://forum.redis.com">Redis Forum</a> or create an issue on <a href="https://github.com/redis-field-engineering/redis-enterprise-kafka">Github</a> where we provide support on a good faith effort basis.]]>
 					</supportSummary>
 					<supportUrl>${project.issueManagement.url}</supportUrl>
 					<confluentControlCenterIntegration>true</confluentControlCenterIntegration>
 					<componentTypes>
-						<componentType>sink</componentType>
 						<componentType>source</componentType>
+						<componentType>sink</componentType>
 					</componentTypes>
 					<tags>
 						<tag>redis</tag>
-						<tag>enterprise</tag>
 					</tags>
 				</configuration>
 				<executions>
@@ -305,11 +303,11 @@
 						</goals>
 						<configuration>
 							<description>Kafka Connect source and sink connectors for Redis
-								Enterprise. It is for Confluent Plaform 5.0 and above</description>
+								(Confluent Plaform 5.x)</description>
 							<version>5.${project.version}</version>
 							<requirements>
-								<requirement>Confluent Platform 5.0+</requirement>
-								<requirement>Redis Enterprise 6.0+</requirement>
+								<requirement>Confluent Platform 5.x</requirement>
+								<requirement>Redis</requirement>
 							</requirements>
 						</configuration>
 					</execution>
@@ -321,11 +319,11 @@
 						</goals>
 						<configuration>
 							<description>Kafka Connect source and sink connectors for Redis
-								Enterprise. It is for Confluent Plaform 6.0 and above</description>
+								(Confluent Plaform 6.0 and above)</description>
 							<version>6.${project.version}</version>
 							<requirements>
 								<requirement>Confluent Platform 6.0+</requirement>
-								<requirement>Redis Enterprise 6.0+</requirement>
+								<requirement>Redis</requirement>
 							</requirements>
 							<excludes>
 								<exclude>io.netty:*</exclude>

run.sh

@@ -8,7 +8,7 @@ if lsof -Pi :6379 -sTCP:LISTEN -t >/dev/null ; then
 fi
 )
 
-echo "Building the Redis Enterprise Kafka Connector"
+echo "Building the Redis Kafka Connector"
 (
 ./mvnw clean package -DskipTests
 mv target/components/packages/redis-redis-enterprise-kafka-6.*.zip target/components/packages/redis-enterprise-kafka.zip
@@ -20,8 +20,8 @@ docker-compose up -d --build
 function clean_up {
     echo -e "\n\nSHUTTING DOWN\n\n"
     curl --output /dev/null -X DELETE http://localhost:8083/connectors/datagen-pageviews || true
-    curl --output /dev/null -X DELETE http://localhost:8083/connectors/redis-enterprise-sink || true
-    curl --output /dev/null -X DELETE http://localhost:8083/connectors/redis-enterprise-source || true
+    curl --output /dev/null -X DELETE http://localhost:8083/connectors/redis-sink || true
+    curl --output /dev/null -X DELETE http://localhost:8083/connectors/redis-source || true
     docker-compose down
     if [ -z "$1" ]
     then
@@ -84,9 +84,9 @@ sleep 5
 
 echo -e "\nAdding Redis Enteprise Kafka Sink Connector for the 'pageviews' topic into the 'pageviews' stream:"
 curl -X POST -H "Content-Type: application/json" --data '
-  {"name": "redis-enterprise-sink",
+  {"name": "redis-sink",
    "config": {
-     "connector.class":"com.redis.kafka.connect.RedisEnterpriseSinkConnector",
+     "connector.class":"com.redis.kafka.connect.RedisSinkConnector",
      "tasks.max":"1",
      "topics":"pageviews",
      "redis.uri":"redis://redis:6379",
@@ -96,11 +96,11 @@ curl -X POST -H "Content-Type: application/json" --data '
 }}' http://localhost:8083/connectors -w "\n"
 
 sleep 2
-echo -e "\nAdding Redis Enterprise Kafka Sink Connector for the 'pageviews' topic into RedisJSON:"
+echo -e "\nAdding Redis Kafka Sink Connector for the 'pageviews' topic into RedisJSON:"
 curl -X POST -H "Content-Type: application/json" --data '
-  {"name": "redis-enterprise-sink-json",
+  {"name": "redis-sink-json",
    "config": {
-     "connector.class":"com.redis.kafka.connect.RedisEnterpriseSinkConnector",
+     "connector.class":"com.redis.kafka.connect.RedisSinkConnector",
      "tasks.max":"1",
      "topics":"pageviews",
      "redis.uri":"redis://redis:6379",
@@ -114,12 +114,12 @@ curl -X POST -H "Content-Type: application/json" --data '
 }}' http://localhost:8083/connectors -w "\n"
 
 sleep 2
-echo -e "\nAdding Redis Enterprise Kafka Source Connector for the 'mystream' stream:"
+echo -e "\nAdding Redis Kafka Source Connector for the 'mystream' stream:"
 curl -X POST -H "Content-Type: application/json" --data '
-  {"name": "redis-enterprise-source",
+  {"name": "redis-source",
    "config": {
      "tasks.max":"1",
-     "connector.class":"com.redis.kafka.connect.RedisEnterpriseSourceConnector",
+     "connector.class":"com.redis.kafka.connect.RedisSourceConnector",
      "redis.uri":"redis://redis:6379",
      "redis.stream.name":"mystream",
      "topic": "mystream"

src/docs/asciidoc/_connect.adoc

@@ -1,7 +1,7 @@
 [[connect]]
 = Connect to Redis
 
-This section provides information on configuring the Redis Enterprise Kafka Source or Sink Connector to connect to Redis Enterprise.
+This section provides information on configuring the Redis Kafka Source or Sink Connector.
 
 == Configuration
 

src/docs/asciidoc/_docker.adoc

@@ -1,10 +1,10 @@
 [[docker]]
 = Quick Start with Docker
 
-This guide provides a hands-on look at the functionality of the Redis Enterprise Kafka Source and Sink Connectors:
+This guide provides a hands-on look at the functionality of the Redis Kafka Source and Sink Connectors:
 
-* The *redis-enterprise-sink* connector reads data from a Kafka topic and writes it to a Redis stream
-* The *redis-enterprise-source* connector reads data from a Redis stream and writes it to a Kafka topic
+* The *redis-sink* connector reads data from a Kafka topic and writes it to a Redis stream
+* The *redis-source* connector reads data from a Redis stream and writes it to a Kafka topic
 
 == Requirements
 
@@ -17,7 +17,6 @@ Clone the https://github.com/{github-owner}/{github-repo}.git[{github-repo}] rep
 [source,console,subs="attributes"]
 ----
 git clone https://github.com/{github-owner}/{github-repo}.git
-cd redis-enterprise-kafka/docker
 ./run.sh
 ----
 
@@ -26,8 +25,8 @@ This will:
 * Run `docker-compose up`
 * Wait for Redis, Kafka, and Kafka Connect to be ready
 * Register the Confluent Datagen Connector
-* Register the Redis Enterprise Kafka Sink Connector
-* Register the Redis Enterprise Kafka Source Connector
+* Register the Redis Kafka Sink Connector
+* Register the Redis Kafka Source Connector
 * Publish some events to Kafka via the Datagen connector
 * Write the events to Redis
 * Send messages to a Redis stream

src/docs/asciidoc/_sink.adoc

@@ -1,8 +1,8 @@
 [[sink]]
 = Sink Connector Guide
-:name: Redis Enterprise Kafka Sink Connector
+:name: Redis Kafka Sink Connector
 
-The {name} consumes records from a Kafka topic and writes the data to a Redis Enterprise database.
+The {name} consumes records from a Kafka topic and writes the data to Redis.
 
 == Features
 

src/docs/asciidoc/_source.adoc

@@ -1,8 +1,8 @@
 [[source]]
 = Source Connector Guide
-:name: Redis Enterprise Kafka Source Connector
+:name: Redis Kafka Source Connector
 
-The {name} reads from a Redis Enterprise stream and publishes messages to a Kafka topic.
+The {name} reads from a Redis stream and publishes messages to a Kafka topic.
 
 == Features
 

src/docs/asciidoc/images/redis-enterprise-kafka-connector.svg renamed to src/docs/asciidoc/images/redis-kafka-connector.svg

@@ -5,14 +5,14 @@
 
 == Introduction
 
-The {project-name} is used to import and export data between Apache Kafka and Redis Enterprise.
+The {project-name} is used to import and export data between Apache Kafka and Redis.
 
-image:images/redis-enterprise-kafka-connector.svg[]
+image:images/redis-kafka-connector.svg[]
 
 This guide provides documentation and usage information across the following topics:
 
 * <<install,Install>>
-* <<connect,Connect to Redis Enterprise>>
+* <<connect,Connect to Redis>>
 * <<sink,Sink Connector>>
 * <<source,Source Connector>>
 * <<docker,Docker Example>>

src/docs/asciidoc/index.adoc

@@ -15,8 +15,8 @@
  */
 package com.redis.kafka.connect;
 
-import com.redis.kafka.connect.sink.RedisEnterpriseSinkConfig;
-import com.redis.kafka.connect.sink.RedisEnterpriseSinkTask;
+import com.redis.kafka.connect.sink.RedisSinkConfig;
+import com.redis.kafka.connect.sink.RedisSinkTask;
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.common.utils.AppInfoParser;
 import org.apache.kafka.connect.connector.Task;
@@ -26,7 +26,7 @@
 import java.util.List;
 import java.util.Map;
 
-public class RedisEnterpriseSinkConnector extends SinkConnector {
+public class RedisSinkConnector extends SinkConnector {
 
     private Map<String, String> props;
 
@@ -37,7 +37,7 @@ public void start(Map<String, String> props) {
 
     @Override
     public Class<? extends Task> taskClass() {
-        return RedisEnterpriseSinkTask.class;
+        return RedisSinkTask.class;
     }
 
     @Override
@@ -51,7 +51,7 @@ public void stop() {
 
     @Override
     public ConfigDef config() {
-        return new RedisEnterpriseSinkConfig.RedisEnterpriseSinkConfigDef();
+        return new RedisSinkConfig.RedisSinkConfigDef();
     }
 
 

src/main/java/com/redis/kafka/connect/RedisEnterpriseSinkConnector.java renamed to src/main/java/com/redis/kafka/connect/RedisSinkConnector.java

@@ -12,8 +12,8 @@
  */
 package com.redis.kafka.connect;
 
-import com.redis.kafka.connect.source.RedisEnterpriseSourceConfig;
-import com.redis.kafka.connect.source.RedisEnterpriseSourceTask;
+import com.redis.kafka.connect.source.RedisSourceConfig;
+import com.redis.kafka.connect.source.RedisSourceTask;
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.common.config.ConfigException;
 import org.apache.kafka.common.utils.AppInfoParser;
@@ -28,29 +28,29 @@
 import java.util.Map;
 import java.util.stream.Collectors;
 
-public class RedisEnterpriseSourceConnector extends SourceConnector {
+public class RedisSourceConnector extends SourceConnector {
 
     private Map<String, String> props;
-    private RedisEnterpriseSourceConfig config;
+    private RedisSourceConfig config;
 
     @Override
     public void start(Map<String, String> props) {
         this.props = props;
         try {
-            this.config = new RedisEnterpriseSourceConfig(props);
+            this.config = new RedisSourceConfig(props);
         } catch (ConfigException configException) {
             throw new ConnectException(configException);
         }
     }
 
     @Override
     public Class<? extends Task> taskClass() {
-        return RedisEnterpriseSourceTask.class;
+        return RedisSourceTask.class;
     }
 
     @Override
     public List<Map<String, String>> taskConfigs(int maxTasks) {
-        if (this.config.getReaderType() == RedisEnterpriseSourceConfig.ReaderType.KEYS) {
+        if (this.config.getReaderType() == RedisSourceConfig.ReaderType.KEYS) {
             // Partition the configs based on channels
             final List<List<String>> partitionedPatterns = ConnectorUtils
                     .groupPartitions(this.config.getKeyPatterns(), Math.min(this.config.getKeyPatterns().size(), maxTasks));
@@ -61,15 +61,15 @@ public List<Map<String, String>> taskConfigs(int maxTasks) {
         List<Map<String, String>> taskConfigs = new ArrayList<>();
         for (int i = 0; i < maxTasks; i++) {
             Map<String, String> taskConfig = new HashMap<>(this.props);
-            taskConfig.put(RedisEnterpriseSourceTask.TASK_ID, Integer.toString(i));
+            taskConfig.put(RedisSourceTask.TASK_ID, Integer.toString(i));
             taskConfigs.add(taskConfig);
         }
         return taskConfigs;
     }
 
     private Map<String, String> taskConfig(List<String> patterns) {
         final Map<String, String> taskConfig = new HashMap<>(this.config.originalsStrings());
-        taskConfig.put(RedisEnterpriseSourceConfig.KEY_PATTERNS_CONFIG, String.join(",", patterns));
+        taskConfig.put(RedisSourceConfig.KEY_PATTERNS_CONFIG, String.join(",", patterns));
         return taskConfig;
     }
 
@@ -79,7 +79,7 @@ public void stop() {
 
     @Override
     public ConfigDef config() {
-        return new RedisEnterpriseSourceConfig.RedisEnterpriseSourceConfigDef();
+        return new RedisSourceConfig.RedisSourceConfigDef();
     }
 
     @Override