[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 <code@tomcooper.dev>

Author: tomncooper

.gitignore

@@ -4,6 +4,7 @@ scalastyle-output.xml
 .classpath
 .idea/*
 !.idea/vcs.xml
+.vscode
 .metadata
 .settings
 .project

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,8 +72,8 @@ public Set<KafkaStream> getAllStreams() {
     @Override
     public Map<String, KafkaStream> describeStreams(Collection<String> streamIds) {
         try {
-            return getAdminClient().describeTopics(new ArrayList<>(streamIds)).all().get().keySet()
-                    .stream()
+            return getAdminClient().describeTopics(new ArrayList<>(streamIds)).allTopicNames().get()
+                    .keySet().stream()
                     .collect(Collectors.toMap(topic -> topic, this::createKafkaStream));
         } catch (InterruptedException | ExecutionException e) {
             throw new RuntimeException("Fetching all streams failed", 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,11 @@ 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";
+    /**
+     * The delivery timeout has to be greater than the request timeout as the latter is part of the
+     * former and this is enforced by a compile time check.
+     */
+    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(
-                                GROUP_ID,
-                                new ListConsumerGroupOffsetsOptions()
-                                        .topicPartitions(
-                                                new ArrayList<>(committedOffsets.keySet())))
+                                Collections.singletonMap(
+                                        GROUP_ID,
+                                        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

@@ -282,8 +282,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;
         }
 

flink-sql-connector-kafka/src/main/resources/META-INF/NOTICE

@@ -6,4 +6,4 @@ The Apache Software Foundation (http://www.apache.org/).
 
 This project bundles the following dependencies under the Apache Software License 2.0. (http://www.apache.org/licenses/LICENSE-2.0.txt)
 
-- org.apache.kafka:kafka-clients:3.9.1
+- org.apache.kafka:kafka-clients:4.0.0