[KafkaIO] Use consumer position and lag to estimate end offsets

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,8 +839,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')
+}

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')
-}
-

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')
-}
-

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')
-}
-

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')
-}
-

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')
+}

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"
+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')
+}

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')
+}

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')
+}

runners/spark/spark_runner.gradle

@@ -287,10 +287,9 @@ 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.kafka_server
   testImplementation library.java.junit
   testImplementation library.java.mockito_core
   testImplementation "org.assertj:assertj-core:3.11.1"

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) {

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")