Merge pull request #288 from lensesio-dev/chore/kafka-4.1

Upgrade to Kafka 4.1.0

Author: stheppi

.github/workflows/build.yml

@@ -191,7 +191,7 @@ jobs:
     strategy:
       matrix:
         module: ${{fromJSON(needs.initiate.outputs.fun_matrix)}}
-        connectImageVersion: [7.3.1, 6.2.2]
+        connectImageVersion: [8.0.2, 8.1.0]
     steps:
       - uses: actions/checkout@v4
       - name: Set up JDK 17

.github/workflows/func-only.yml

@@ -12,7 +12,7 @@ jobs:
     strategy:
       matrix:
         module: ["http"]
-        connectImageVersion: [ 7.3.1, 6.2.2 ]
+        connectImageVersion: [ 8.0.2, 8.1.0 ]
 
     steps:
       - uses: actions/checkout@v4

build.sbt

@@ -35,11 +35,9 @@ lazy val subProjects: Seq[Project] = Seq(
   ftp,
   `gcp-storage`,
   http,
-  influxdb,
   jms,
   mongodb,
   mqtt,
-  redis,
 )
 
 lazy val subProjectsRefs: Seq[ProjectReference] = subProjects.map(projectToLocalProject)
@@ -399,27 +397,6 @@ lazy val http = (project in file("kafka-connect-http"))
   .configureFunctionalTests()
   .enablePlugins(PackPlugin, ProtocPlugin)
 
-lazy val influxdb = (project in file("kafka-connect-influxdb"))
-  .dependsOn(common)
-  .dependsOn(`sql-common`)
-  .settings(
-    settings ++
-      Seq(
-        name := "kafka-connect-influxdb",
-        description := "Kafka Connect compatible connectors to move data between Kafka and popular data stores",
-        libraryDependencies ++= baseDeps ++ kafkaConnectInfluxDbDeps,
-        publish / skip := true,
-        packExcludeJars := Seq(
-          "scala-.*\\.jar",
-          "zookeeper-.*\\.jar",
-        ),
-      ),
-  )
-  .configureAssembly(true)
-  .configureMavenDescriptor()
-  .configureTests(baseTestDeps)
-  .enablePlugins(PackPlugin)
-
 lazy val jms = (project in file("kafka-connect-jms"))
   .dependsOn(common)
   .dependsOn(`sql-common`)
@@ -439,7 +416,7 @@ lazy val jms = (project in file("kafka-connect-jms"))
         Compile / PB.targets := Seq(
           PB.gens.java(Versions.googleProtobufVersion) -> (Test / sourceManaged).value,
         ),
-        Compile / PB.protocExecutable := file(Process("which protoc").!!.trim),
+        Compile / PB.protocVersion := Versions.googleProtobufVersion,
       ),
   )
   .configureAssembly(true)
@@ -520,31 +497,6 @@ lazy val mongodb = (project in file("kafka-connect-mongodb"))
   .configureFunctionalTests()
   .enablePlugins(PackPlugin)
 
-lazy val redis = (project in file("kafka-connect-redis"))
-  .dependsOn(common)
-  .dependsOn(`sql-common`)
-  .dependsOn(`test-common` % "fun->compile")
-  .settings(
-    settings ++
-      Seq(
-        name := "kafka-connect-redis",
-        description := "Kafka Connect compatible connectors to move data between Kafka and popular data stores",
-        libraryDependencies ++= baseDeps ++ kafkaConnectRedisDeps,
-        publish / skip := true,
-        FunctionalTest / baseDirectory := (LocalRootProject / baseDirectory).value,
-        packExcludeJars := Seq(
-          "scala-.*\\.jar",
-          "zookeeper-.*\\.jar",
-        ),
-      ),
-  )
-  .configureAssembly(true)
-  .configureMavenDescriptor()
-  .configureTests(baseTestDeps ++ Seq(gson))
-  .configureIntegrationTests(kafkaConnectRedisTestDeps)
-  .configureFunctionalTests()
-  .enablePlugins(PackPlugin)
-
 lazy val `test-common` = (project in file("test-common"))
   .settings(
     settings ++

java-connectors/build.gradle

@@ -28,18 +28,19 @@ allprojects {
     apply plugin: 'com.diffplug.spotless'
 
     java {
-        setSourceCompatibility(JavaVersion.VERSION_11)
-        setTargetCompatibility(JavaVersion.VERSION_11)
+        // Use Java 17 to match Kafka 4.1.0+ artifacts which require JVM 17
+        setSourceCompatibility(JavaVersion.VERSION_17)
+        setTargetCompatibility(JavaVersion.VERSION_17)
     }
 
     ext {
         //DEPENDENCY VERSIONS
         lombokVersion = '1.18.32'
-        kafkaVersion = '3.9.1'
+        kafkaVersion = '4.1.0'
         logbackVersion = '1.4.14'
         jUnitVersion = '5.9.1'
         mockitoJupiterVersion = '5.10.0'
-        apacheToConfluentVersionAxis = ["2.8.1": "6.2.2", "3.3.0": "7.3.1"]
+        apacheToConfluentVersionAxis = ["4.0.0": "8.0.2", "4.1.0": "8.1.0"]
         caffeineVersion = '3.1.8'
         cyclopsVersion = '10.4.1'
         bouncyCastleVersion = "1.78.1"

java-connectors/kcbq-connector/build.gradle

@@ -11,7 +11,7 @@ project(":kcbq-connector") {
         googleAuthVersion = '0.21.1'
         googleCloudVersion = '2.10.9'
         googleCloudStorageVersion = '1.113.4'
-        googleProtobufVersion = '3.25.5'
+
         jacksonVersion = '2.16.2'
         kafkaVersion = '2.6.3'
         kafkaScalaVersion = '2.12'

kafka-connect-aws-s3/src/it/scala/io/lenses/streamreactor/connect/aws/s3/source/S3SourceTaskTest.scala

@@ -16,6 +16,10 @@ import io.lenses.streamreactor.connect.cloud.common.config.FormatOptions
 import io.lenses.streamreactor.connect.cloud.common.model.location.CloudLocation
 import io.lenses.streamreactor.connect.cloud.common.model.location.CloudLocationValidator
 import io.lenses.streamreactor.connect.cloud.common.source.config.CloudSourceSettingsKeys
+import org.apache.kafka.common.MetricName
+import org.apache.kafka.common.metrics.MetricValueProvider
+import org.apache.kafka.common.metrics.PluginMetrics
+import org.apache.kafka.common.metrics.Sensor
 import org.apache.kafka.connect.source.SourceTaskContext
 import org.apache.kafka.connect.storage.OffsetStorageReader
 import org.scalatest.BeforeAndAfter
@@ -29,6 +33,11 @@ import org.scalatest.time.SpanSugar.convertIntToGrainOfTime
 import java.util
 import scala.jdk.CollectionConverters.ListHasAsScala
 import scala.jdk.CollectionConverters.MapHasAsJava
+import org.apache.kafka.common.metrics.Metrics
+import org.apache.kafka.common.metrics.MetricConfig
+import org.apache.kafka.common.utils.Time
+import org.apache.kafka.common.metrics.Sensor.RecordingLevel
+
 object S3SourceTaskTest {
 
   val formats = Table(
@@ -214,6 +223,33 @@ class S3SourceTaskTest
         val task = new S3SourceTask()
 
         val context = new SourceTaskContext {
+          override def pluginMetrics(): PluginMetrics = new PluginMetrics {
+            private val metricConfig    = new MetricConfig()
+            private val metricsRegistry = new Metrics(metricConfig, Time.SYSTEM)
+            private val parents: Array[Sensor] = Array.empty
+            override def metricName(
+              name:        String,
+              description: String,
+              tags:        util.LinkedHashMap[String, String],
+            ): MetricName =
+              new MetricName(name, "s3-source-task-metrics", description, tags)
+
+            override def addMetric(metricName: MetricName, metricValueProvider: MetricValueProvider[_]): Unit = {}
+
+            override def removeMetric(metricName: MetricName): Unit = {}
+
+            override def addSensor(name: String): Sensor =
+              metricsRegistry.sensor(
+                name,
+                metricConfig,
+                Long.MaxValue,
+                RecordingLevel.INFO,
+                parents: _*,
+              )
+
+            override def removeSensor(name: String): Unit =
+              metricsRegistry.removeSensor(name)
+          }
           override def configs(): util.Map[String, String] = Map.empty[String, String].asJava
 
           override def offsetStorageReader(): OffsetStorageReader = new OffsetStorageReader {