From a6d831a3d9be0aa210354294e27b7d7d51f68157 Mon Sep 17 00:00:00 2001
From: Steven van Rossum <sjvanrossum@google.com>
Date: Sun, 16 Nov 2025 23:30:10 +0000
Subject: [PATCH 1/7] Remove support for Kafka clients older than 3.9.2

---
 .../beam/gradle/BeamModulePlugin.groovy       |  5 ++--
 examples/java/build.gradle                    |  4 ++--
 runners/spark/spark_runner.gradle             |  4 +---
 sdks/java/io/kafka/build.gradle               | 12 +++-------
 sdks/java/io/kafka/kafka-231/build.gradle     | 24 -------------------
 sdks/java/io/kafka/kafka-241/build.gradle     | 24 -------------------
 sdks/java/io/kafka/kafka-251/build.gradle     | 24 -------------------
 sdks/java/io/kafka/kafka-282/build.gradle     | 24 -------------------
 sdks/java/io/kafka/kafka-312/build.gradle     | 24 -------------------
 sdks/java/io/kafka/kafka-390/build.gradle     | 24 -------------------
 .../{kafka-201 => kafka-392}/build.gradle     |  6 ++---
 sdks/java/testing/kafka-service/build.gradle  |  3 ++-
 settings.gradle.kts                           | 16 ++-----------
 13 files changed, 16 insertions(+), 178 deletions(-)
 delete mode 100644 sdks/java/io/kafka/kafka-231/build.gradle
 delete mode 100644 sdks/java/io/kafka/kafka-241/build.gradle
 delete mode 100644 sdks/java/io/kafka/kafka-251/build.gradle
 delete mode 100644 sdks/java/io/kafka/kafka-282/build.gradle
 delete mode 100644 sdks/java/io/kafka/kafka-312/build.gradle
 delete mode 100644 sdks/java/io/kafka/kafka-390/build.gradle
 rename sdks/java/io/kafka/{kafka-201 => kafka-392}/build.gradle (90%)

diff --git a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
index 3843958f0f3e..d7ccfcf580db 100644
--- a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
+++ b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
@@ -628,7 +628,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
@@ -839,7 +839,8 @@ 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",
         log4j                                       : "log4j:log4j:1.2.17",
         log4j_over_slf4j                            : "org.slf4j:log4j-over-slf4j:$slf4j_version",
diff --git a/examples/java/build.gradle b/examples/java/build.gradle
index 34d884b97780..84ace728362e 100644
--- a/examples/java/build.gradle
+++ b/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")
diff --git a/runners/spark/spark_runner.gradle b/runners/spark/spark_runner.gradle
index 1da044ab7b52..ee5817571f9d 100644
--- a/runners/spark/spark_runner.gradle
+++ b/runners/spark/spark_runner.gradle
@@ -287,9 +287,7 @@ dependencies {
   testImplementation project(path: ":sdks:java:extensions:avro", configuration: "testRuntimeMigration")
   testImplementation project(":sdks:java:harness")
   testImplementation library.java.avro
-  // kafka_2.13 artifacts were first published in 2.5.0; use a later version for Scala 2.13
-  def kafka_version = (spark_scala_version == '2.13') ? '2.8.0' : '2.4.1'
-  testImplementation "org.apache.kafka:kafka_$spark_scala_version:$kafka_version"
+  testImplementation (spark_scala_version == '2.13' ? library.java.kafka_scala_2_13 : library.java.kafka_scala_2_12)
   testImplementation library.java.kafka_clients
   testImplementation library.java.junit
   testImplementation library.java.mockito_core
diff --git a/sdks/java/io/kafka/build.gradle b/sdks/java/io/kafka/build.gradle
index 13969f7a9aed..0d28469eae56 100644
--- a/sdks/java/io/kafka/build.gradle
+++ b/sdks/java/io/kafka/build.gradle
@@ -36,13 +36,7 @@ ext {
 }
 
 def kafkaVersions = [
-    '201': "2.0.1",
-    '231': "2.3.1",
-    '241': "2.4.1",
-    '251': "2.5.1",
-    '282': "2.8.2",
-    '312': "3.1.2",
-    '390': "3.9.0",
+    '392': "3.9.2",
 ]
 
 kafkaVersions.each{k,v -> configurations.create("kafkaVersion$k")}
@@ -63,8 +57,8 @@ dependencies {
   if (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
   }
   testImplementation library.java.kafka_clients
   testImplementation project(path: ":runners:core-java")
diff --git a/sdks/java/io/kafka/kafka-231/build.gradle b/sdks/java/io/kafka/kafka-231/build.gradle
deleted file mode 100644
index 712158dcd3ae..000000000000
--- a/sdks/java/io/kafka/kafka-231/build.gradle
+++ /dev/null
@@ -1,24 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * License); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an AS IS BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-project.ext {
-    delimited="2.3.1"
-    undelimited="231"
-    sdfCompatible=true
-}
-
-apply from: "../kafka-integration-test.gradle"
\ No newline at end of file
diff --git a/sdks/java/io/kafka/kafka-241/build.gradle b/sdks/java/io/kafka/kafka-241/build.gradle
deleted file mode 100644
index c0ac7df674b5..000000000000
--- a/sdks/java/io/kafka/kafka-241/build.gradle
+++ /dev/null
@@ -1,24 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * License); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an AS IS BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-project.ext {
-    delimited="2.4.1"
-    undelimited="241"
-    sdfCompatible=true
-}
-
-apply from: "../kafka-integration-test.gradle"
\ No newline at end of file
diff --git a/sdks/java/io/kafka/kafka-251/build.gradle b/sdks/java/io/kafka/kafka-251/build.gradle
deleted file mode 100644
index 4de9f97a738a..000000000000
--- a/sdks/java/io/kafka/kafka-251/build.gradle
+++ /dev/null
@@ -1,24 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * License); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an AS IS BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-project.ext {
-    delimited="2.5.1"
-    undelimited="251"
-    sdfCompatible=true
-}
-
-apply from: "../kafka-integration-test.gradle"
\ No newline at end of file
diff --git a/sdks/java/io/kafka/kafka-282/build.gradle b/sdks/java/io/kafka/kafka-282/build.gradle
deleted file mode 100644
index b754d93077e5..000000000000
--- a/sdks/java/io/kafka/kafka-282/build.gradle
+++ /dev/null
@@ -1,24 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * License); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an AS IS BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-project.ext {
-    delimited="2.8.2"
-    undelimited="282"
-    sdfCompatible=true
-}
-
-apply from: "../kafka-integration-test.gradle"
\ No newline at end of file
diff --git a/sdks/java/io/kafka/kafka-312/build.gradle b/sdks/java/io/kafka/kafka-312/build.gradle
deleted file mode 100644
index af2ad3717b65..000000000000
--- a/sdks/java/io/kafka/kafka-312/build.gradle
+++ /dev/null
@@ -1,24 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * License); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an AS IS BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-project.ext {
-    delimited="3.1.2"
-    undelimited="312"
-    sdfCompatible=true
-}
-
-apply from: "../kafka-integration-test.gradle"
\ No newline at end of file
diff --git a/sdks/java/io/kafka/kafka-390/build.gradle b/sdks/java/io/kafka/kafka-390/build.gradle
deleted file mode 100644
index 8c8821386265..000000000000
--- a/sdks/java/io/kafka/kafka-390/build.gradle
+++ /dev/null
@@ -1,24 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * License); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an AS IS BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-project.ext {
-    delimited="3.9.0"
-    undelimited="390"
-    sdfCompatible=true
-}
-
-apply from: "../kafka-integration-test.gradle"
\ No newline at end of file
diff --git a/sdks/java/io/kafka/kafka-201/build.gradle b/sdks/java/io/kafka/kafka-392/build.gradle
similarity index 90%
rename from sdks/java/io/kafka/kafka-201/build.gradle
rename to sdks/java/io/kafka/kafka-392/build.gradle
index a26ca4ac19cf..3df1ebeb69ad 100644
--- a/sdks/java/io/kafka/kafka-201/build.gradle
+++ b/sdks/java/io/kafka/kafka-392/build.gradle
@@ -16,9 +16,9 @@
  * limitations under the License.
  */
 project.ext {
-    delimited="2.0.1"
-    undelimited="201"
+    delimited="3.9.2"
+    undelimited="392"
     sdfCompatible=true
 }
 
-apply from: "../kafka-integration-test.gradle"
\ No newline at end of file
+apply from: "../kafka-integration-test.gradle"
diff --git a/sdks/java/testing/kafka-service/build.gradle b/sdks/java/testing/kafka-service/build.gradle
index abd186f98b1d..1148ebe6572c 100644
--- a/sdks/java/testing/kafka-service/build.gradle
+++ b/sdks/java/testing/kafka-service/build.gradle
@@ -27,7 +27,8 @@ ext.summary = """Self-contained Kafka service for testing IO transforms."""
 
 
 dependencies {
-  testImplementation library.java.kafka
+  testImplementation library.java.kafka_scala_2_12
+  testImplementation library.java.kafka_server
   testImplementation "org.apache.zookeeper:zookeeper:3.5.6"
   testRuntimeOnly library.java.slf4j_log4j12
 }
diff --git a/settings.gradle.kts b/settings.gradle.kts
index d9dbbf9021ef..f71f6de9d167 100644
--- a/settings.gradle.kts
+++ b/settings.gradle.kts
@@ -364,20 +364,8 @@ project(":beam-test-gha").projectDir = file(".github")
 include("beam-validate-runner")
 project(":beam-validate-runner").projectDir = file(".test-infra/validate-runner")
 include("com.google.api.gax.batching")
-include("sdks:java:io:kafka:kafka-390")
-findProject(":sdks:java:io:kafka:kafka-390")?.name = "kafka-390"
-include("sdks:java:io:kafka:kafka-312")
-findProject(":sdks:java:io:kafka:kafka-312")?.name = "kafka-312"
-include("sdks:java:io:kafka:kafka-282")
-findProject(":sdks:java:io:kafka:kafka-282")?.name = "kafka-282"
-include("sdks:java:io:kafka:kafka-251")
-findProject(":sdks:java:io:kafka:kafka-251")?.name = "kafka-251"
-include("sdks:java:io:kafka:kafka-241")
-findProject(":sdks:java:io:kafka:kafka-241")?.name = "kafka-241"
-include("sdks:java:io:kafka:kafka-231")
-findProject(":sdks:java:io:kafka:kafka-231")?.name = "kafka-231"
-include("sdks:java:io:kafka:kafka-201")
-findProject(":sdks:java:io:kafka:kafka-201")?.name = "kafka-201"
+include("sdks:java:io:kafka:kafka-392")
+findProject(":sdks:java:io:kafka:kafka-392")?.name = "kafka-392"
 include("sdks:java:managed")
 findProject(":sdks:java:managed")?.name = "managed"
 include("sdks:java:io:iceberg")

From b191394c02a14c782c9bc246d6d362ce63d5756e Mon Sep 17 00:00:00 2001
From: Steven van Rossum <sjvanrossum@google.com>
Date: Mon, 13 Jul 2026 23:09:11 +0200
Subject: [PATCH 2/7] Resolve capability conflicts for Flink 1.x, Flink 2.0,
 Spark 3 and Spark 4

---
 examples/java/common.gradle                | 1 +
 runners/flink/1.19/build.gradle            | 6 ++++++
 runners/flink/1.19/job-server/build.gradle | 6 ++++++
 runners/flink/1.20/build.gradle            | 6 ++++++
 runners/flink/1.20/job-server/build.gradle | 6 ++++++
 runners/flink/2.0/build.gradle             | 6 ++++++
 runners/flink/2.0/job-server/build.gradle  | 6 ++++++
 runners/flink/2.1/build.gradle             | 7 -------
 runners/flink/2.1/job-server/build.gradle  | 7 -------
 runners/flink/2.2/build.gradle             | 7 -------
 runners/flink/2.2/job-server/build.gradle  | 7 -------
 runners/spark/3/build.gradle               | 6 ++++++
 runners/spark/3/job-server/build.gradle    | 8 +++++++-
 runners/spark/4/build.gradle               | 5 +++++
 runners/spark/4/job-server/build.gradle    | 6 ++++++
 sdks/java/testing/nexmark/build.gradle     | 7 ++++---
 sdks/java/testing/tpcds/build.gradle       | 7 ++++---
 17 files changed, 69 insertions(+), 35 deletions(-)

diff --git a/examples/java/common.gradle b/examples/java/common.gradle
index f32667d733fb..91d07aff76ba 100644
--- a/examples/java/common.gradle
+++ b/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 {
diff --git a/runners/flink/1.19/build.gradle b/runners/flink/1.19/build.gradle
index 1545da258477..4116b817eec4 100644
--- a/runners/flink/1.19/build.gradle
+++ b/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')
+}
diff --git a/runners/flink/1.19/job-server/build.gradle b/runners/flink/1.19/job-server/build.gradle
index 332f04e08ceb..c9e09a5de8c4 100644
--- a/runners/flink/1.19/job-server/build.gradle
+++ b/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')
+}
diff --git a/runners/flink/1.20/build.gradle b/runners/flink/1.20/build.gradle
index 4c148321ed49..86a1cb7ef54d 100644
--- a/runners/flink/1.20/build.gradle
+++ b/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')
+}
diff --git a/runners/flink/1.20/job-server/build.gradle b/runners/flink/1.20/job-server/build.gradle
index e5fdd1febf92..9f129b2c021c 100644
--- a/runners/flink/1.20/job-server/build.gradle
+++ b/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')
+}
diff --git a/runners/flink/2.0/build.gradle b/runners/flink/2.0/build.gradle
index 490bc593f40c..4e034173fb93 100644
--- a/runners/flink/2.0/build.gradle
+++ b/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')
+}
diff --git a/runners/flink/2.0/job-server/build.gradle b/runners/flink/2.0/job-server/build.gradle
index 6d068f839491..ebac0a8e039c 100644
--- a/runners/flink/2.0/job-server/build.gradle
+++ b/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')
+}
diff --git a/runners/flink/2.1/build.gradle b/runners/flink/2.1/build.gradle
index 1e0d565b50d9..32c71eb96483 100644
--- a/runners/flink/2.1/build.gradle
+++ b/runners/flink/2.1/build.gradle
@@ -41,10 +41,3 @@ project.ext {
 
 // Load the main build script which contains all build logic.
 apply from: "../flink_runner.gradle"
-
-// Flink 2.1 uses at.yawk.lz4:lz4-java instead of org.lz4:lz4-java
-// Explicitly prefer Flink's at.yawk.lz4 version to resolve capability conflict
-configurations.all {
-  resolveCapabilitiesConflict(it, 'org.lz4:lz4-java', 'at.yawk.lz4')
-}
-
diff --git a/runners/flink/2.1/job-server/build.gradle b/runners/flink/2.1/job-server/build.gradle
index 0910fef11208..3135abcb19b7 100644
--- a/runners/flink/2.1/job-server/build.gradle
+++ b/runners/flink/2.1/job-server/build.gradle
@@ -29,10 +29,3 @@ project.ext {
 
 // Load the main build script which contains all build logic.
 apply from: "$basePath/flink_job_server.gradle"
-
-// Flink 2.1 uses at.yawk.lz4:lz4-java instead of org.lz4:lz4-java
-// Explicitly prefer Flink's at.yawk.lz4 version to resolve capability conflict
-configurations.all {
-  resolveCapabilitiesConflict(it, 'org.lz4:lz4-java', 'at.yawk.lz4')
-}
-
diff --git a/runners/flink/2.2/build.gradle b/runners/flink/2.2/build.gradle
index 0321dcf42d10..1e1d7fdd5e75 100644
--- a/runners/flink/2.2/build.gradle
+++ b/runners/flink/2.2/build.gradle
@@ -56,10 +56,3 @@ project.ext {
 
 // Load the main build script which contains all build logic.
 apply from: "../flink_runner.gradle"
-
-// Flink 2.2 uses at.yawk.lz4:lz4-java instead of org.lz4:lz4-java
-// Explicitly prefer Flink's at.yawk.lz4 version to resolve capability conflict
-configurations.all {
-  resolveCapabilitiesConflict(it, 'org.lz4:lz4-java', 'at.yawk.lz4')
-}
-
diff --git a/runners/flink/2.2/job-server/build.gradle b/runners/flink/2.2/job-server/build.gradle
index f116f5a1dcbf..527245329211 100644
--- a/runners/flink/2.2/job-server/build.gradle
+++ b/runners/flink/2.2/job-server/build.gradle
@@ -29,10 +29,3 @@ project.ext {
 
 // Load the main build script which contains all build logic.
 apply from: "$basePath/flink_job_server.gradle"
-
-// Flink 2.2 uses at.yawk.lz4:lz4-java instead of org.lz4:lz4-java
-// Explicitly prefer Flink's at.yawk.lz4 version to resolve capability conflict
-configurations.all {
-  resolveCapabilitiesConflict(it, 'org.lz4:lz4-java', 'at.yawk.lz4')
-}
-
diff --git a/runners/spark/3/build.gradle b/runners/spark/3/build.gradle
index c2c492a6b2e9..274fa45d2f75 100644
--- a/runners/spark/3/build.gradle
+++ b/runners/spark/3/build.gradle
@@ -88,3 +88,9 @@ tasks.register("sparkVersionsTest") {
   group = "Verification"
   dependsOn sparkVersions.collect{k,v -> "sparkVersion${k}Test"}
 }
+
+// Spark 3 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')
+}
diff --git a/runners/spark/3/job-server/build.gradle b/runners/spark/3/job-server/build.gradle
index 68bb8d9a10e1..59165e80be58 100644
--- a/runners/spark/3/job-server/build.gradle
+++ b/runners/spark/3/job-server/build.gradle
@@ -28,4 +28,10 @@ project.ext {
 }
 
 // Load the main build script which contains all build logic.
-apply from: "$basePath/spark_job_server.gradle"
\ No newline at end of file
+apply from: "$basePath/spark_job_server.gradle"
+
+// Spark 3 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')
+}
diff --git a/runners/spark/4/build.gradle b/runners/spark/4/build.gradle
index ec1af8df38a4..f2746b06158e 100644
--- a/runners/spark/4/build.gradle
+++ b/runners/spark/4/build.gradle
@@ -61,3 +61,8 @@ tasks.named("copyTestSourceOverrides") {
   exclude "**/translation/streaming/**"
 }
 
+// Spark 4 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')
+}
diff --git a/runners/spark/4/job-server/build.gradle b/runners/spark/4/job-server/build.gradle
index 598cf3b4913a..3154d92ea325 100644
--- a/runners/spark/4/job-server/build.gradle
+++ b/runners/spark/4/job-server/build.gradle
@@ -29,3 +29,9 @@ project.ext {
 
 // Load the main build script which contains all build logic.
 apply from: "$basePath/spark_job_server.gradle"
+
+// Spark 4 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')
+}
diff --git a/sdks/java/testing/nexmark/build.gradle b/sdks/java/testing/nexmark/build.gradle
index bd917a3935ad..21137efcb955 100644
--- a/sdks/java/testing/nexmark/build.gradle
+++ b/sdks/java/testing/nexmark/build.gradle
@@ -39,13 +39,13 @@ def nexmarkRunnerDependency = project.findProperty(nexmarkRunnerProperty)
 def nexmarkRunnerVersionProperty = "nexmark.runner.version"
 def nexmarkRunnerVersion = project.findProperty(nexmarkRunnerVersionProperty)
 def isSparkRunner = nexmarkRunnerDependency.startsWith(":runners:spark:")
-def isFlink2 = nexmarkRunnerDependency.startsWith(":runners:flink:") && {
+def gtFlink20 = nexmarkRunnerDependency.startsWith(":runners:flink:") && {
   def version = nexmarkRunnerDependency.substring(":runners:flink:".length())
   try {
     def parts = version.split('\\.')
     def major = parts[0].toInteger()
     def minor = parts.length > 1 ? parts[1].toInteger() : 0
-    return (major > 2) || (major == 2 && minor >= 1)
+    return (major == 2 && minor > 0)
   } catch (Exception e) {
     return false
   }
@@ -103,7 +103,7 @@ dependencies {
   gradleRun project(path: nexmarkRunnerDependency, configuration: runnerConfiguration)
 }
 
-if (isFlink2) {
+if (!gtFlink20) {
   resolveCapabilitiesConflict(configurations.gradleRun, 'org.lz4:lz4-java', 'at.yawk.lz4')
 }
 
@@ -112,6 +112,7 @@ if (isSparkRunner) {
     // Using Spark runner causes a StackOverflowError if slf4j-jdk14 is on the classpath
     exclude group: "org.slf4j", module: "slf4j-jdk14"
   }
+  resolveCapabilitiesConflict(configurations.gradleRun, 'org.lz4:lz4-java', 'at.yawk.lz4')
 }
 
 def getNexmarkArgs = {
diff --git a/sdks/java/testing/tpcds/build.gradle b/sdks/java/testing/tpcds/build.gradle
index 6a05df96ab20..13c16e17bea8 100644
--- a/sdks/java/testing/tpcds/build.gradle
+++ b/sdks/java/testing/tpcds/build.gradle
@@ -34,13 +34,13 @@ def tpcdsRunnerProperty = "tpcds.runner"
 def tpcdsRunnerDependency = project.findProperty(tpcdsRunnerProperty)
         ?: ":runners:direct-java"
 def isSpark = tpcdsRunnerDependency.startsWith(":runners:spark:")
-def isFlink2 = tpcdsRunnerDependency.startsWith(":runners:flink:") && {
+def gtFlink20 = tpcdsRunnerDependency.startsWith(":runners:flink:") && {
   def version = tpcdsRunnerDependency.substring(":runners:flink:".length())
   try {
     def parts = version.split('\\.')
     def major = parts[0].toInteger()
     def minor = parts.length > 1 ? parts[1].toInteger() : 0
-    return (major > 2) || (major == 2 && minor >= 1)
+    return (major == 2 && minor > 0)
   } catch (Exception e) {
     return false
   }
@@ -94,7 +94,7 @@ dependencies {
     gradleRun project(path: tpcdsRunnerDependency, configuration: runnerConfiguration)
 }
 
-if (isFlink2) {
+if (!gtFlink20) {
     resolveCapabilitiesConflict(configurations.gradleRun, 'org.lz4:lz4-java', 'at.yawk.lz4')
 }
 
@@ -102,6 +102,7 @@ if (isSpark) {
     configurations.gradleRun {
       exclude group: "org.slf4j", module: "slf4j-jdk14"
     }
+    resolveCapabilitiesConflict(configurations.gradleRun, 'org.lz4:lz4-java', 'at.yawk.lz4')
 }
 
 // Execute the TPC-DS queries or suites via Gradle.

From bac0de664f3c3acf21723f9008221e5a80515209 Mon Sep 17 00:00:00 2001
From: Steven van Rossum <sjvanrossum@google.com>
Date: Tue, 14 Jul 2026 01:01:44 +0200
Subject: [PATCH 3/7] Resolve capability conflicts for load tests and
 watermarks

---
 sdks/java/testing/load-tests/build.gradle | 16 +++++++++++++++
 sdks/java/testing/watermarks/build.gradle | 25 ++++++++++++++++++++++-
 2 files changed, 40 insertions(+), 1 deletion(-)

diff --git a/sdks/java/testing/load-tests/build.gradle b/sdks/java/testing/load-tests/build.gradle
index c74c7301db74..2eb2eb2362cf 100644
--- a/sdks/java/testing/load-tests/build.gradle
+++ b/sdks/java/testing/load-tests/build.gradle
@@ -40,6 +40,17 @@ def runnerDependency = (project.hasProperty(runnerProperty)
 def loadTestRunnerVersionProperty = "runner.version"
 def loadTestRunnerVersion = project.findProperty(loadTestRunnerVersionProperty)
 def isSparkRunner = runnerDependency.startsWith(":runners:spark:")
+def gtFlink20 = runnerDependency.startsWith(":runners:flink:") && {
+  def version = runnerDependency.substring(":runners:flink:".length())
+  try {
+    def parts = version.split('\\.')
+    def major = parts[0].toInteger()
+    def minor = parts.length > 1 ? parts[1].toInteger() : 0
+    return (major == 2 && minor > 0)
+  } catch (Exception e) {
+    return false
+  }
+}()
 def isDataflowRunner = ":runners:google-cloud-dataflow-java".equals(runnerDependency)
 def isDataflowRunnerV2 = isDataflowRunner && "V2".equals(loadTestRunnerVersion)
 def runnerConfiguration = ":runners:direct-java".equals(runnerDependency) ? "shadow" : null
@@ -89,11 +100,16 @@ dependencies {
   gradleRun project(path: runnerDependency, configuration: runnerConfiguration)
 }
 
+if (!gtFlink20) {
+  resolveCapabilitiesConflict(configurations.gradleRun, 'org.lz4:lz4-java', 'at.yawk.lz4')
+}
+
 if (isSparkRunner) {
   configurations.gradleRun {
     // Using Spark runner causes a StackOverflowError if slf4j-jdk14 is on the classpath
     exclude group: "org.slf4j", module: "slf4j-jdk14"
   }
+  resolveCapabilitiesConflict(configurations.gradleRun, 'org.lz4:lz4-java', 'at.yawk.lz4')
 }
 
 def getLoadTestArgs = {
diff --git a/sdks/java/testing/watermarks/build.gradle b/sdks/java/testing/watermarks/build.gradle
index ca774815467a..74955685b7c9 100644
--- a/sdks/java/testing/watermarks/build.gradle
+++ b/sdks/java/testing/watermarks/build.gradle
@@ -38,7 +38,18 @@ def runnerProperty = "runner"
 def runnerDependency = (project.hasProperty(runnerProperty)
         ? project.getProperty(runnerProperty)
         : ":runners:direct-java")
-
+def isSparkRunner = runnerDependency.startsWith(":runners:spark:")
+def gtFlink20 = runnerDependency.startsWith(":runners:flink:") && {
+  def version = runnerDependency.substring(":runners:flink:".length())
+  try {
+    def parts = version.split('\\.')
+    def major = parts[0].toInteger()
+    def minor = parts.length > 1 ? parts[1].toInteger() : 0
+    return (major == 2 && minor > 0)
+  } catch (Exception e) {
+    return false
+  }
+}()
 def isDataflowRunner = ":runners:google-cloud-dataflow-java".equals(runnerDependency)
 def runnerConfiguration = ":runners:direct-java".equals(runnerDependency) ? "shadow" : null
 
@@ -74,6 +85,18 @@ dependencies {
   gradleRun project(path: runnerDependency, configuration: runnerConfiguration)
 }
 
+if (!gtFlink20) {
+  resolveCapabilitiesConflict(configurations.gradleRun, 'org.lz4:lz4-java', 'at.yawk.lz4')
+}
+
+if (isSparkRunner) {
+  configurations.gradleRun {
+    // Using Spark runner causes a StackOverflowError if slf4j-jdk14 is on the classpath
+    exclude group: "org.slf4j", module: "slf4j-jdk14"
+  }
+  resolveCapabilitiesConflict(configurations.gradleRun, 'org.lz4:lz4-java', 'at.yawk.lz4')
+}
+
 task run(type: JavaExec) {
   def loadTestArgs = project.findProperty(loadTestArgsProperty) ?: ""
 

From 11a3295397c59c9552e9a0152ea5f7decefd71ff Mon Sep 17 00:00:00 2001
From: Steven van Rossum <sjvanrossum@google.com>
Date: Thu, 27 Nov 2025 23:19:44 +0000
Subject: [PATCH 4/7] Replace obsolete signature of overridden method close in
 mock consumer

---
 .../org/apache/beam/sdk/io/kafka/KafkaCommitOffsetTest.java   | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaCommitOffsetTest.java b/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaCommitOffsetTest.java
index c16e25510ab8..b64f5edabe76 100644
--- a/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaCommitOffsetTest.java
+++ b/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaCommitOffsetTest.java
@@ -17,11 +17,11 @@
  */
 package org.apache.beam.sdk.io.kafka;
 
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.TimeUnit;
 import org.apache.beam.sdk.coders.CannotProvideCoderException;
 import org.apache.beam.sdk.coders.KvCoder;
 import org.apache.beam.sdk.coders.StringUtf8Coder;
@@ -236,7 +236,7 @@ public synchronized void commitSync(Map<TopicPartition, OffsetAndMetadata> offse
     }
 
     @Override
-    public synchronized void close(long timeout, TimeUnit unit) {
+    public synchronized void close(Duration timeout) {
       // Ignore closing since we're using a single consumer.
     }
   }

From 022d4ecd411e647d6adf665fe13a5f2f6217aad5 Mon Sep 17 00:00:00 2001
From: Steven van Rossum <sjvanrossum@google.com>
Date: Sat, 20 Dec 2025 14:07:59 +0000
Subject: [PATCH 5/7] Replace obsolete class KafkaServerStartable with
 KafkaServer

---
 .../apache/beam/gradle/BeamModulePlugin.groovy  |  1 +
 runners/spark/spark_runner.gradle               |  1 +
 .../streaming/utils/EmbeddedKafkaCluster.java   | 17 +++++++++++------
 .../beam/sdk/testing/kafka/LocalKafka.java      | 11 ++++++++---
 4 files changed, 21 insertions(+), 9 deletions(-)

diff --git a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
index d7ccfcf580db..8a3c63ee2d19 100644
--- a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
+++ b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
@@ -842,6 +842,7 @@ class BeamModulePlugin implements Plugin<Project> {
         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",
diff --git a/runners/spark/spark_runner.gradle b/runners/spark/spark_runner.gradle
index ee5817571f9d..77da3d36db92 100644
--- a/runners/spark/spark_runner.gradle
+++ b/runners/spark/spark_runner.gradle
@@ -289,6 +289,7 @@ dependencies {
   testImplementation library.java.avro
   testImplementation (spark_scala_version == '2.13' ? library.java.kafka_scala_2_13 : library.java.kafka_scala_2_12)
   testImplementation library.java.kafka_clients
+  testImplementation library.java.kafka_server
   testImplementation library.java.junit
   testImplementation library.java.mockito_core
   testImplementation "org.assertj:assertj-core:3.11.1"
diff --git a/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/streaming/utils/EmbeddedKafkaCluster.java b/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/streaming/utils/EmbeddedKafkaCluster.java
index df5646fed590..3acbf664e2cf 100644
--- a/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/streaming/utils/EmbeddedKafkaCluster.java
+++ b/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/streaming/utils/EmbeddedKafkaCluster.java
@@ -29,7 +29,7 @@
 import java.util.Properties;
 import java.util.Random;
 import kafka.server.KafkaConfig;
-import kafka.server.KafkaServerStartable;
+import kafka.server.KafkaServer;
 import org.apache.zookeeper.server.NIOServerCnxnFactory;
 import org.apache.zookeeper.server.ServerCnxnFactory;
 import org.apache.zookeeper.server.ZooKeeperServer;
@@ -47,7 +47,7 @@ public class EmbeddedKafkaCluster {
 
   private final String brokerList;
 
-  private final List<KafkaServerStartable> brokers;
+  private final List<KafkaServer> brokers;
   private final List<File> logDirs;
 
   private EmbeddedKafkaCluster(String zkConnection) {
@@ -114,15 +114,20 @@ public void startup() {
       properties.setProperty("offsets.topic.replication.factor", "1");
       properties.setProperty("log.flush.interval.messages", String.valueOf(1));
 
-      KafkaServerStartable broker = startBroker(properties);
+      KafkaServer broker = startBroker(properties);
 
       brokers.add(broker);
       logDirs.add(logDir);
     }
   }
 
-  private static KafkaServerStartable startBroker(Properties props) {
-    KafkaServerStartable server = new KafkaServerStartable(new KafkaConfig(props));
+  private static KafkaServer startBroker(Properties props) {
+    KafkaServer server =
+        new KafkaServer(
+            new KafkaConfig(props),
+            KafkaServer.$lessinit$greater$default$2(),
+            KafkaServer.$lessinit$greater$default$3(),
+            KafkaServer.$lessinit$greater$default$4());
     server.startup();
     return server;
   }
@@ -148,7 +153,7 @@ public String getZkConnection() {
 
   @SuppressWarnings("Slf4jDoNotLogMessageOfExceptionExplicitly")
   public void shutdown() {
-    for (KafkaServerStartable broker : brokers) {
+    for (KafkaServer broker : brokers) {
       try {
         broker.shutdown();
       } catch (Exception e) {
diff --git a/sdks/java/testing/kafka-service/src/test/java/org/apache/beam/sdk/testing/kafka/LocalKafka.java b/sdks/java/testing/kafka-service/src/test/java/org/apache/beam/sdk/testing/kafka/LocalKafka.java
index 71ec61a3e41e..dc88f3fe0a5a 100644
--- a/sdks/java/testing/kafka-service/src/test/java/org/apache/beam/sdk/testing/kafka/LocalKafka.java
+++ b/sdks/java/testing/kafka-service/src/test/java/org/apache/beam/sdk/testing/kafka/LocalKafka.java
@@ -20,10 +20,10 @@
 import java.nio.file.Files;
 import java.util.Properties;
 import kafka.server.KafkaConfig;
-import kafka.server.KafkaServerStartable;
+import kafka.server.KafkaServer;
 
 public class LocalKafka {
-  private final KafkaServerStartable server;
+  private final KafkaServer server;
 
   LocalKafka(int kafkaPort, int zookeeperPort) throws Exception {
     Properties kafkaProperties = new Properties();
@@ -31,7 +31,12 @@ public class LocalKafka {
     kafkaProperties.setProperty("zookeeper.connect", String.format("localhost:%s", zookeeperPort));
     kafkaProperties.setProperty("offsets.topic.replication.factor", "1");
     kafkaProperties.setProperty("log.dir", Files.createTempDirectory("kafka-log-").toString());
-    server = new KafkaServerStartable(KafkaConfig.fromProps(kafkaProperties));
+    server =
+        new KafkaServer(
+            KafkaConfig.fromProps(kafkaProperties),
+            KafkaServer.$lessinit$greater$default$2(),
+            KafkaServer.$lessinit$greater$default$3(),
+            KafkaServer.$lessinit$greater$default$4());
   }
 
   public void start() {

From 664890dc26483217a2215aa0deb4846e45912731 Mon Sep 17 00:00:00 2001
From: Steven van Rossum <sjvanrossum@google.com>
Date: Sun, 21 Dec 2025 22:33:59 +0000
Subject: [PATCH 6/7] Fix type ambiguity of method argument in test

---
 .../org/apache/beam/it/kafka/KafkaResourceManagerTest.java     | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/it/kafka/src/test/java/org/apache/beam/it/kafka/KafkaResourceManagerTest.java b/it/kafka/src/test/java/org/apache/beam/it/kafka/KafkaResourceManagerTest.java
index 8c870815efc4..f25bca06c226 100644
--- a/it/kafka/src/test/java/org/apache/beam/it/kafka/KafkaResourceManagerTest.java
+++ b/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

From f7dbcc2f2f7bcbb87874bcef196527041c63e417 Mon Sep 17 00:00:00 2001
From: Steven van Rossum <sjvanrossum@google.com>
Date: Fri, 10 Jul 2026 14:34:20 +0200
Subject: [PATCH 7/7] Handle Kafka and executor timeouts the same

---
 .../sdk/io/kafka/KafkaUnboundedReader.java    | 28 +++++++++++--------
 1 file changed, 17 insertions(+), 11 deletions(-)

diff --git a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java
index c5dc5c408fe2..c3aa6f5418bc 100644
--- a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java
+++ b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java
@@ -30,6 +30,7 @@
 import java.util.NoSuchElementException;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
@@ -114,17 +115,22 @@ public boolean start() throws IOException {
       try {
         Duration timeout = resolveDefaultApiTimeout(spec);
         future.get(timeout.getMillis(), TimeUnit.MILLISECONDS);
-      } catch (TimeoutException e) {
-        consumer.wakeup(); // This unblocks consumer stuck on network I/O.
-        // Likely reason : Kafka servers are configured to advertise internal ips, but
-        // those ips are not accessible from workers outside.
-        String msg =
-            String.format(
-                "%s: Timeout while initializing partition '%s'. "
-                    + "Kafka client may not be able to connect to servers.",
-                this, pState.topicPartition);
-        LOG.error("{}", msg);
-        throw new IOException(msg);
+      } catch (TimeoutException | ExecutionException e) {
+        if (e instanceof TimeoutException
+            || e.getCause() instanceof org.apache.kafka.common.errors.TimeoutException) {
+          // TODO: Find out if manually waking up was only relevant for legacy Kafka clients.
+          consumer.wakeup(); // This unblocks consumer stuck on network I/O.
+          // Likely reason : Kafka servers are configured to advertise internal ips, but
+          // those ips are not accessible from workers outside.
+          String msg =
+              String.format(
+                  "%s: Timeout while initializing partition '%s'. "
+                      + "Kafka client may not be able to connect to servers.",
+                  this, pState.topicPartition);
+          LOG.error("{}", msg);
+          throw new IOException(msg);
+        }
+        throw new IOException(e);
       } catch (Exception e) {
         throw new IOException(e);
       }