[KafkaIO] Remove build support for Kafka clients before 3.9.2 (#39284)

* Remove support for Kafka clients older than 3.9.2

* Resolve capability conflicts for Flink 1.x, Flink 2.0, Spark 3 and Spark 4

* Resolve capability conflicts for load tests and watermarks

* Replace obsolete signature of overridden method close in mock consumer

* Replace obsolete class KafkaServerStartable with KafkaServer

* Fix type ambiguity of method argument in test

* Handle Kafka and executor timeouts the same

Author: sjvanrossum

buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy

@@ -639,7 +639,7 @@ class BeamModulePlugin implements Plugin<Project> {
     def jaxb_api_version = "2.3.3"
     def jsr305_version = "3.0.2"
     def everit_json_version = "1.14.2"
-    def kafka_version = "2.4.1"
+    def kafka_version = "3.9.2"
     def log4j2_version = "2.25.4"
     def nemo_version = "0.1"
     // [bomupgrader] determined by: io.grpc:grpc-netty, consistent with: google_cloud_platform_libraries_bom
@@ -850,8 +850,10 @@ class BeamModulePlugin implements Plugin<Project> {
         jupiter_api                                 : "org.junit.jupiter:junit-jupiter-api:$jupiter_version",
         jupiter_engine                              : "org.junit.jupiter:junit-jupiter-engine:$jupiter_version",
         jupiter_params                              : "org.junit.jupiter:junit-jupiter-params:$jupiter_version",
-        kafka                                       : "org.apache.kafka:kafka_2.11:$kafka_version",
+        kafka_scala_2_12                            : "org.apache.kafka:kafka_2.12:$kafka_version",
+        kafka_scala_2_13                            : "org.apache.kafka:kafka_2.13:$kafka_version",
         kafka_clients                               : "org.apache.kafka:kafka-clients:$kafka_version",
+        kafka_server                                : "org.apache.kafka:kafka-server:$kafka_version",
         log4j                                       : "log4j:log4j:1.2.17",
         log4j_over_slf4j                            : "org.slf4j:log4j-over-slf4j:$slf4j_version",
         log4j2_api                                  : "org.apache.logging.log4j:log4j-api:$log4j2_version",

examples/java/build.gradle

@@ -44,8 +44,8 @@ dependencies {
   if (project.findProperty('testJavaVersion') == '21' || JavaVersion.current().compareTo(JavaVersion.VERSION_21) >= 0) {
     // this dependency is a provided dependency for kafka-avro-serializer. It is not needed to compile with Java<=17
     // but needed for compile only under Java21, specifically, required for extending from AbstractKafkaAvroDeserializer
-    compileOnly library.java.kafka
-    permitUnusedDeclared library.java.kafka
+    compileOnly library.java.kafka_scala_2_12
+    permitUnusedDeclared library.java.kafka_scala_2_12
   }
   implementation library.java.kafka_clients
   implementation project(path: ":sdks:java:core", configuration: "shadow")

examples/java/common.gradle

@@ -36,6 +36,7 @@ configurations.sparkRunnerPreCommit {
   exclude group: "org.slf4j", module: "slf4j-jdk14"
 }
 resolveCapabilitiesConflict(configurations.flinkRunnerPreCommit, 'org.lz4:lz4-java', 'at.yawk.lz4')
+resolveCapabilitiesConflict(configurations.sparkRunnerPreCommit, 'org.lz4:lz4-java', 'at.yawk.lz4')
 
 
 dependencies {

it/kafka/src/test/java/org/apache/beam/it/kafka/KafkaResourceManagerTest.java

@@ -166,7 +166,8 @@ public void testCleanupShouldDropNonStaticTopic() throws IOException {
     KafkaResourceManager tm = new KafkaResourceManager(kafkaClient, container, builder);
 
     tm.cleanupAll();
-    verify(kafkaClient).deleteTopics(argThat(list -> list.size() == numTopics));
+    verify(kafkaClient)
+        .deleteTopics(argThat((Collection<String> list) -> list.size() == numTopics));
   }
 
   @Test

runners/flink/1.19/build.gradle

@@ -23,3 +23,9 @@ project.ext {
 
 // Load the main build script which contains all build logic.
 apply from: "../flink_runner.gradle"
+
+// Flink 1.19 uses org.lz4:lz4-java instead of at.yawk.lz4:lz4-java
+// Explicitly prefer at.yawk.lz4 candidates to resolve capability conflict
+configurations.all {
+  resolveCapabilitiesConflict(it, 'org.lz4:lz4-java', 'at.yawk.lz4')
+}

runners/flink/1.19/job-server/build.gradle

@@ -29,3 +29,9 @@ project.ext {
 
 // Load the main build script which contains all build logic.
 apply from: "$basePath/flink_job_server.gradle"
+
+// Flink 1.19 uses org.lz4:lz4-java instead of at.yawk.lz4:lz4-java
+// Explicitly prefer at.yawk.lz4 candidates to resolve capability conflict
+configurations.all {
+  resolveCapabilitiesConflict(it, 'org.lz4:lz4-java', 'at.yawk.lz4')
+}

runners/flink/1.20/build.gradle

@@ -23,3 +23,9 @@ project.ext {
 
 // Load the main build script which contains all build logic.
 apply from: "../flink_runner.gradle"
+
+// Flink 1.20 uses org.lz4:lz4-java instead of at.yawk.lz4:lz4-java
+// Explicitly prefer at.yawk.lz4 candidates to resolve capability conflict
+configurations.all {
+  resolveCapabilitiesConflict(it, 'org.lz4:lz4-java', 'at.yawk.lz4')
+}

runners/flink/1.20/job-server/build.gradle

@@ -29,3 +29,9 @@ project.ext {
 
 // Load the main build script which contains all build logic.
 apply from: "$basePath/flink_job_server.gradle"
+
+// Flink 1.20 uses org.lz4:lz4-java instead of at.yawk.lz4:lz4-java
+// Explicitly prefer at.yawk.lz4 candidates to resolve capability conflict
+configurations.all {
+  resolveCapabilitiesConflict(it, 'org.lz4:lz4-java', 'at.yawk.lz4')
+}

runners/flink/2.0/build.gradle

@@ -41,3 +41,9 @@ project.ext {
 
 // Load the main build script which contains all build logic.
 apply from: "../flink_runner.gradle"
+
+// Flink 2.0 uses org.lz4:lz4-java instead of at.yawk.lz4:lz4-java
+// Explicitly prefer at.yawk.lz4 candidates to resolve capability conflict
+configurations.all {
+  resolveCapabilitiesConflict(it, 'org.lz4:lz4-java', 'at.yawk.lz4')
+}

runners/flink/2.0/job-server/build.gradle

@@ -29,3 +29,9 @@ project.ext {
 
 // Load the main build script which contains all build logic.
 apply from: "$basePath/flink_job_server.gradle"
+
+// Flink 2.0 uses org.lz4:lz4-java instead of at.yawk.lz4:lz4-java
+// Explicitly prefer at.yawk.lz4 candidates to resolve capability conflict
+configurations.all {
+  resolveCapabilitiesConflict(it, 'org.lz4:lz4-java', 'at.yawk.lz4')
+}