GH-3814: Upgrade Apache Kafka client dependencies to version 4.0.0

Fixes: #3814
Issue link: #3814

This commit upgrades the Apache Kafka client to 4.0.0 with the following changes:

- Upgrade Kafka client dependencies to 4.0.0 in build.gradle
- Remove ZooKeeper-based broker implementation as Kafka 4.0 fully transitions to KRaft mode
  - Delete EmbeddedKafkaZKBroker class and related tests
  - Remove ZooKeeper dependency
  - Refactor EmbeddedKafkaRule to use KRaft exclusively
  - Update KafkaClusterTestKit imports to use new packages

- Update ConsumerRecords constructor calls to include the new required Map parameter
- Add implementations for new Producer interface methods:
  - registerMetricForSubscription
  - unregisterMetricFromSubscription

- Update KafkaStreamBrancher to use new split() and branch() methods
- Remove deprecated partitioner classes from runtime hints
- Remove deprecated sendOffsetsToTransaction method that used String consumerGroupId
- Update BrokerAddress to use org.apache.kafka.server.network.BrokerEndPoint
- Update DeserializationExceptionHandler to use new ErrorHandlerContext

The commit also includes test modifications to address limitations with static port assignments in KRaft mode
and adjustments to replication factors in tests.

Signed-off-by: Soby Chacko <[email protected]>

* Addressing PR review

---------

Signed-off-by: Soby Chacko <[email protected]>

Author: sobychacko

Diff for: build.gradle

@@ -59,7 +59,7 @@ ext {
 	jaywayJsonPathVersion = '2.9.0'
 	junit4Version = '4.13.2'
 	junitJupiterVersion = '5.11.4'
-	kafkaVersion = '3.8.1'
+	kafkaVersion = '4.0.0'
 	kotlinCoroutinesVersion = '1.10.1'
 	log4jVersion = '2.24.3'
 	micrometerDocsVersion = '1.0.4'
@@ -355,16 +355,14 @@ project ('spring-kafka-test') {
 	description = 'Spring Kafka Test Support'
 
 	dependencies {
+		api "org.apache.logging.log4j:log4j-slf4j-impl:$log4jVersion"
 		api 'org.springframework:spring-context'
 		api 'org.springframework:spring-test'
 		api "org.springframework.retry:spring-retry:$springRetryVersion"
 
-		api ("org.apache.zookeeper:zookeeper:$zookeeperVersion") {
-			exclude group: 'org.slf4j', module: 'slf4j-log4j12'
-			exclude group: 'log4j'
-		}
 		api "org.apache.kafka:kafka-clients:$kafkaVersion:test"
 		api "org.apache.kafka:kafka-server:$kafkaVersion"
+		api "org.apache.kafka:kafka-test-common-runtime:$kafkaVersion"
 		api "org.apache.kafka:kafka-metadata:$kafkaVersion"
 		api "org.apache.kafka:kafka-server-common:$kafkaVersion"
 		api "org.apache.kafka:kafka-server-common:$kafkaVersion:test"

Diff for: spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBrokerFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2024 the original author or authors.
+ * Copyright 2024-2025 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -63,12 +63,8 @@ public static EmbeddedKafkaBroker create(EmbeddedKafka embeddedKafka, Function<S
 						.toArray(String[]::new);
 
 		EmbeddedKafkaBroker embeddedKafkaBroker;
-		if (embeddedKafka.kraft()) {
-			embeddedKafkaBroker = kraftBroker(embeddedKafka, topics);
-		}
-		else {
-			embeddedKafkaBroker = zkBroker(embeddedKafka, topics);
-		}
+		embeddedKafkaBroker = kraftBroker(embeddedKafka, topics);
+
 		int[] ports = setupPorts(embeddedKafka);
 
 		embeddedKafkaBroker.kafkaPorts(ports)
@@ -133,13 +129,6 @@ private static EmbeddedKafkaBroker kraftBroker(EmbeddedKafka embedded, String[]
 		return new EmbeddedKafkaKraftBroker(embedded.count(), embedded.partitions(), topics);
 	}
 
-	private static EmbeddedKafkaBroker zkBroker(EmbeddedKafka embedded, String[] topics) {
-		return new EmbeddedKafkaZKBroker(embedded.count(), embedded.controlledShutdown(), embedded.partitions(), topics)
-				.zkPort(embedded.zookeeperPort())
-				.zkConnectionTimeout(embedded.zkConnectionTimeout())
-				.zkSessionTimeout(embedded.zkSessionTimeout());
-	}
-
 	private EmbeddedKafkaBrokerFactory() {
 	}
 

Diff for: spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaKraftBroker.java

@@ -41,8 +41,6 @@
 import java.util.stream.Collectors;
 
 import kafka.server.KafkaConfig;
-import kafka.testkit.KafkaClusterTestKit;
-import kafka.testkit.TestKitNodes;
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.clients.admin.AdminClient;
@@ -53,6 +51,8 @@
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.test.KafkaClusterTestKit;
+import org.apache.kafka.common.test.TestKitNodes;
 import org.apache.kafka.common.utils.Exit;
 import org.apache.kafka.common.utils.Utils;
 import org.jspecify.annotations.Nullable;