[FLINK-37583] Upgrade to Kafka 4.0.0 client.

Note that this update will make the connector incompatible with Kafka
clusters running Kafka version 2.0 and older.

Signed-off-by: Thomas Cooper <[email protected]>

Author: tomncooper

docs/content.zh/docs/connectors/datastream/kafka.md

@@ -33,7 +33,7 @@ Flink 提供了 [Apache Kafka](https://kafka.apache.org) 连接器使用精确
 
 Apache Flink 集成了通用的 Kafka 连接器，它会尽力与 Kafka client 的最新版本保持同步。
 该连接器使用的 Kafka client 版本可能会在 Flink 版本之间发生变化。
-当前 Kafka client 向后兼容 0.10.0 或更高版本的 Kafka broker。
+当前 Kafka client 向后兼容 2.1.0 或更高版本的 Kafka broker。
 有关 Kafka 兼容性的更多细节，请参考 [Kafka 官方文档](https://kafka.apache.org/protocol.html#protocol_compatibility)。
 
 {{< connector_artifact flink-connector-kafka kafka >}}

docs/content/docs/connectors/datastream/kafka.md

@@ -33,7 +33,7 @@ Flink provides an [Apache Kafka](https://kafka.apache.org) connector for reading
 
 Apache Flink ships with a universal Kafka connector which attempts to track the latest version of the Kafka client.
 The version of the client it uses may change between Flink releases.
-Modern Kafka clients are backwards compatible with broker versions 0.10.0 or later.
+Modern Kafka clients are backwards compatible with broker versions 2.1.0 or later.
 For details on Kafka compatibility, please refer to the official [Kafka documentation](https://kafka.apache.org/protocol.html#protocol_compatibility).
 
 {{< connector_artifact flink-connector-kafka kafka >}}

flink-connector-kafka/.gitignore

@@ -0,0 +1 @@
+/latest_offset_resume_topic*

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/metadata/SingleClusterTopicMetadataService.java

@@ -72,7 +72,7 @@ public Set<KafkaStream> getAllStreams() {
     @Override
     public Map<String, KafkaStream> describeStreams(Collection<String> streamIds) {
         try {
-            return getAdminClient().describeTopics(new ArrayList<>(streamIds)).all().get().keySet()
+            return getAdminClient().describeTopics(new ArrayList<>(streamIds)).allTopicNames().get().keySet()
                     .stream()
                     .collect(Collectors.toMap(topic -> topic, this::createKafkaStream));
         } catch (InterruptedException | ExecutionException e) {

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumerator.java

@@ -32,7 +32,7 @@
 
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.KafkaAdminClient;
-import org.apache.kafka.clients.admin.ListConsumerGroupOffsetsOptions;
+import org.apache.kafka.clients.admin.ListConsumerGroupOffsetsSpec;
 import org.apache.kafka.clients.admin.ListOffsetsResult;
 import org.apache.kafka.clients.admin.OffsetSpec;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
@@ -533,12 +533,11 @@ public PartitionOffsetsRetrieverImpl(AdminClient adminClient, String groupId) {
 
         @Override
         public Map<TopicPartition, Long> committedOffsets(Collection<TopicPartition> partitions) {
-            ListConsumerGroupOffsetsOptions options =
-                    new ListConsumerGroupOffsetsOptions()
-                            .topicPartitions(new ArrayList<>(partitions));
+            ListConsumerGroupOffsetsSpec offsetsSpec =
+                    new ListConsumerGroupOffsetsSpec().topicPartitions(partitions);
             try {
                 return adminClient
-                        .listConsumerGroupOffsets(groupId, options)
+                        .listConsumerGroupOffsets(Collections.singletonMap(groupId, offsetsSpec))
                         .partitionsToOffsetAndMetadata()
                         .thenApply(
                                 result -> {

flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterFaultToleranceITCase.java

@@ -40,7 +40,7 @@ public class KafkaWriterFaultToleranceITCase extends KafkaWriterTestBase {
     private static final String INIT_KAFKA_RETRIES = "0";
     private static final String INIT_KAFKA_REQUEST_TIMEOUT_MS = "1000";
     private static final String INIT_KAFKA_MAX_BLOCK_MS = "1000";
-    private static final String INIT_KAFKA_DELIVERY_TIMEOUT_MS = "1000";
+    private static final String INIT_KAFKA_DELIVERY_TIMEOUT_MS = "1500";
 
     @RegisterExtension
     public static final MiniClusterExtension MINI_CLUSTER_RESOURCE =

flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/testutils/KafkaSourceTestEnv.java

@@ -22,7 +22,7 @@
 import org.apache.flink.streaming.connectors.kafka.KafkaTestBase;
 
 import org.apache.kafka.clients.admin.AdminClient;
-import org.apache.kafka.clients.admin.ListConsumerGroupOffsetsOptions;
+import org.apache.kafka.clients.admin.ListConsumerGroupOffsetsSpec;
 import org.apache.kafka.clients.admin.RecordsToDelete;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
@@ -36,6 +36,7 @@
 
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -238,10 +239,10 @@ public static void setupCommittedOffsets(String topic)
         Map<TopicPartition, OffsetAndMetadata> toVerify =
                 adminClient
                         .listConsumerGroupOffsets(
+                                Collections.singletonMap(
                                 GROUP_ID,
-                                new ListConsumerGroupOffsetsOptions()
-                                        .topicPartitions(
-                                                new ArrayList<>(committedOffsets.keySet())))
+                                new ListConsumerGroupOffsetsSpec()
+                                        .topicPartitions(committedOffsets.keySet())))
                         .partitionsToOffsetAndMetadata()
                         .get();
         assertThat(toVerify).as("The offsets are not committed").isEqualTo(committedOffsets);

flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java

@@ -292,8 +292,13 @@ public KafkaOffsetHandlerImpl() {
 
         @Override
         public Long getCommittedOffset(String topicName, int partition) {
-            OffsetAndMetadata committed =
-                    offsetClient.committed(new TopicPartition(topicName, partition));
+            TopicPartition topicPartition = new TopicPartition(topicName, partition);
+            Map<TopicPartition, OffsetAndMetadata> committedMap =
+                    offsetClient.committed(Collections.singleton(topicPartition));
+            if (committedMap == null) {
+                return null;
+            }
+            OffsetAndMetadata committed = committedMap.get(topicPartition);
             return (committed != null) ? committed.offset() : null;
         }
 

pom.xml

@@ -27,7 +27,7 @@ under the License.
 
     <modelVersion>4.0.0</modelVersion>
 
-    <artifactId>flink-connector-kafka-parent</artifactId>
+	<artifactId>flink-connector-kafka-parent</artifactId>
     <version>4.0-SNAPSHOT</version>
     <name>Flink : Connectors : Kafka : Parent</name>
     <packaging>pom</packaging>
@@ -50,7 +50,7 @@ under the License.
 
     <properties>
         <flink.version>2.0.0</flink.version>
-        <kafka.version>3.9.0</kafka.version>
+        <kafka.version>4.0.0</kafka.version>
         <confluent.version>7.8.2</confluent.version>
 
         <jackson-bom.version>2.16.2</jackson-bom.version>
@@ -64,8 +64,8 @@ under the License.
         <byte-buddy.version>1.12.10</byte-buddy.version>
         <commons-cli.version>1.5.0</commons-cli.version>
         <scala.binary.version>2.12</scala.binary.version>
-        <scala-reflect.version>2.12.19</scala-reflect.version>
-        <scala-library.version>2.12.19</scala-library.version>
+        <scala-reflect.version>2.12.20</scala-reflect.version>
+        <scala-library.version>2.12.20</scala-library.version>
         <snappy-java.version>1.1.10.5</snappy-java.version>
         <avro.version>1.11.4</avro.version>
         <guava.version>32.1.2-jre</guava.version>
@@ -74,7 +74,7 @@ under the License.
         <japicmp.referenceVersion>1.17.0</japicmp.referenceVersion>
 
         <slf4j.version>1.7.36</slf4j.version>
-        <log4j.version>2.17.1</log4j.version>
+        <log4j.version>2.24.3</log4j.version>
 
         <flink.parent.artifactId>flink-connector-kafka-parent</flink.parent.artifactId>