support Kafka 3.6 (#25)

Author: rashtao

.circleci/continue_config.yml

@@ -61,7 +61,7 @@ jobs:
         default: '172.28.0.1:8529,172.28.0.1:8539,172.28.0.1:8549'
       kafka-version:
         type: 'string'
-        default: '3.5'
+        default: '3.6'
     environment:
       STARTER_MODE: <<parameters.topology>>
       DOCKER_IMAGE: docker.io/arangodb/arangodb:<<parameters.arango-version>>
@@ -216,6 +216,7 @@ workflows:
           matrix:
             parameters:
               kafka-version:
+                - '3.6'
                 - '3.5'
                 - '3.4'
                 - '3.3'

demo/docker-compose.yml

@@ -2,7 +2,7 @@ version: "2"
 
 services:
   kafka-1:
-    image: docker.io/bitnami/kafka:3.5
+    image: docker.io/bitnami/kafka:3.6
     ports:
       - '19092:19092'
     environment:
@@ -19,7 +19,7 @@ services:
       KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT,CONTROLLER:PLAINTEXT
 
   kafka-2:
-    image: docker.io/bitnami/kafka:3.5
+    image: docker.io/bitnami/kafka:3.6
     ports:
       - '29092:29092'
     environment:
@@ -36,7 +36,7 @@ services:
       KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT,CONTROLLER:PLAINTEXT
 
   kafka-3:
-    image: docker.io/bitnami/kafka:3.5
+    image: docker.io/bitnami/kafka:3.6
     ports:
       - '39092:39092'
     environment:
@@ -53,7 +53,7 @@ services:
       KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT,CONTROLLER:PLAINTEXT
 
   kafka-connect-1:
-    image: docker.io/bitnami/kafka:3.5
+    image: docker.io/bitnami/kafka:3.6
     depends_on:
       - kafka-1
       - kafka-2
@@ -68,7 +68,7 @@ services:
     command: /opt/bitnami/kafka/bin/connect-distributed.sh /tmp/connect-distributed-1.properties
 
   kafka-connect-2:
-    image: docker.io/bitnami/kafka:3.5
+    image: docker.io/bitnami/kafka:3.6
     depends_on:
       - kafka-1
       - kafka-2

docker/start_schema_registry.sh

@@ -1,6 +1,6 @@
 #!/bin/bash
 
-DOCKER_IMAGE=docker.io/confluentinc/cp-schema-registry:7.4.0
+DOCKER_IMAGE=docker.io/confluentinc/cp-schema-registry:7.5.2
 docker pull $DOCKER_IMAGE
 
 KAFKA_BOOTSTRAP_SERVERS=PLAINTEXT://kafka-1:9092,PLAINTEXT://kafka-2:9092,PLAINTEXT://kafka-3:9092

pom.xml

@@ -60,8 +60,8 @@
         <maven.compiler.source>8</maven.compiler.source>
         <maven.compiler.target>8</maven.compiler.target>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-        <kafka.version>3.4.1</kafka.version>
-        <confluent.version>7.4.0</confluent.version>
+        <kafka.version>3.6.0</kafka.version>
+        <confluent.version>7.5.2</confluent.version>
         <arangodb.version>7.3.0</arangodb.version>
         <jackson.version>2.13.5</jackson.version>
         <slf4j.version>1.7.36</slf4j.version>
@@ -269,6 +269,12 @@
                 <version>1.14.5</version>
                 <scope>test</scope>
             </dependency>
+            <dependency>
+                <groupId>org.apache.commons</groupId>
+                <artifactId>commons-compress</artifactId>
+                <version>1.22</version>
+                <scope>test</scope>
+            </dependency>
         </dependencies>
     </dependencyManagement>
 

src/connect-standalone/java/spi/StandaloneKafkaConnectDeployment.java

@@ -28,13 +28,15 @@
 import org.apache.kafka.connect.runtime.Herder;
 import org.apache.kafka.connect.runtime.Worker;
 import org.apache.kafka.connect.runtime.isolation.Plugins;
-import org.apache.kafka.connect.runtime.rest.RestServer;
+import org.apache.kafka.connect.runtime.rest.ConnectRestServer;
+import org.apache.kafka.connect.runtime.rest.RestClient;
 import org.apache.kafka.connect.runtime.standalone.StandaloneConfig;
 import org.apache.kafka.connect.runtime.standalone.StandaloneHerder;
 import org.apache.kafka.connect.storage.MemoryOffsetBackingStore;
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Set;
 
 public class StandaloneKafkaConnectDeployment extends KafkaConnectDeployment {
 
@@ -74,19 +76,20 @@ public void start() {
 
         Plugins plugins = new Plugins(workerProps);
         StandaloneConfig config = new StandaloneConfig(workerProps);
-
-        AllConnectorClientConfigOverridePolicy allConnectorClientConfigOverridePolicy =
-                new AllConnectorClientConfigOverridePolicy();
-
-        Worker worker = new Worker(
-                workerId, time, plugins, config, new MemoryOffsetBackingStore(),
-                allConnectorClientConfigOverridePolicy);
-        Herder herder = new StandaloneHerder(worker, "cluster-id", allConnectorClientConfigOverridePolicy);
-
-        RestServer rest = new RestServer(config, null);
-        rest.initializeServer();
-
-        new Connect(herder, rest).start();
+        AllConnectorClientConfigOverridePolicy allowOverride = new AllConnectorClientConfigOverridePolicy();
+        MemoryOffsetBackingStore offsetStore = new MemoryOffsetBackingStore() {
+            @Override
+            public Set<Map<String, Object>> connectorPartitions(String connectorName) {
+                throw new UnsupportedOperationException();
+            }
+        };
+        Worker worker = new Worker(workerId, time, plugins, config, offsetStore, allowOverride);
+        Herder herder = new StandaloneHerder(worker, "cluster-id", allowOverride);
+        RestClient restClient = new RestClient(config);
+        ConnectRestServer restServer = new ConnectRestServer(null, restClient, workerProps);
+        restServer.initializeServer();
+
+        new Connect(herder, restServer).start();
     }
 
 }