GH-1966: Upgrade Kafka Clients to 3.0.0

Resolves #1966

* Update spring-kafka-docs/src/main/asciidoc/kafka.adoc

Author: garyrussell

build.gradle

@@ -59,7 +59,7 @@ ext {
 	jaywayJsonPathVersion = '2.4.0'
 	junit4Version = '4.13.2'
 	junitJupiterVersion = '5.7.2'
-	kafkaVersion = '2.8.1'
+	kafkaVersion = '3.0.0'
 	log4jVersion = '2.14.1'
 	micrometerVersion = '1.8.0-M3'
 	mockitoVersion = '3.11.2'
@@ -69,6 +69,7 @@ ext {
 	springDataVersion = '2021.1.0-M3'
 	springRetryVersion = '1.3.1'
 	springVersion = '5.3.10'
+	zookeeperVersion = '3.6.3'
 
 	idPrefix = 'kafka'
 }
@@ -341,8 +342,9 @@ project ('spring-kafka-test') {
 			exclude group: 'org.springframework'
 		}
 
+		api "org.apache.zookeeper:zookeeper:$zookeeperVersion"
 		api "org.apache.kafka:kafka-clients:$kafkaVersion:test"
-		api "org.apache.kafka:kafka-streams:$kafkaVersion"
+		api "org.apache.kafka:kafka-metadata:$kafkaVersion"
 		api "org.apache.kafka:kafka-streams-test-utils:$kafkaVersion"
 		api "org.apache.kafka:kafka_$scalaVersion:$kafkaVersion"
 		api "org.apache.kafka:kafka_$scalaVersion:$kafkaVersion:test"

spring-kafka-docs/src/main/asciidoc/appendix.adoc

@@ -1,3 +1,4 @@
+////
 [[update-deps]]
 == Override Spring Boot Dependencies
 
@@ -82,6 +83,7 @@ dependencies {
 ====
 
 The test scope dependencies are only needed if you are using the embedded Kafka broker in tests.
+////
 
 [appendix]
 [[history]]

spring-kafka-docs/src/main/asciidoc/kafka.adoc

@@ -3555,11 +3555,10 @@ Starting with version 2.6, the default `EOSMode` is `V2`.
 
 To configure the container to use mode `ALPHA`, set the container property `EOSMode` to `ALPHA`, to revert to the previous behavior.
 
-IMPORTANT: With `V2`, your brokers must be version 2.5 or later, however with `kafka-clients` version 2.6, the producer will automatically fall back to `ALPHA` if the broker does not support `BETA`.
-The `DefaultKafkaProducerFactory` is configured to enable that behavior.
-If your brokers are earlier than 2.5, be sure to leave the `DefaultKafkaProducerFactory` `producerPerConsumerPartition` set to `true` and, if you are using a batch listener, you should set `subBatchPerPartition` to `true`.
+IMPORTANT: With `V2` (default), your brokers must be version 2.5 or later; `kafka-clients` version 3.0, the producer will no longer fall back to `V1`; if the broker does not support `V2`, an exception is thrown.
+If your brokers are earlier than 2.5, you must set the `EOSMode` to `V1`, leave the `DefaultKafkaProducerFactory` `producerPerConsumerPartition` set to `true` and, if you are using a batch listener, you should set `subBatchPerPartition` to `true`.
 
-When your brokers are upgraded to 2.5 or later, the producer will automatically use mode `V2`, but the number of producers will remain as before.
+When your brokers are upgraded to 2.5 or later, you should switch the mode to `V2`, but the number of producers will remain as before.
 You can then do a rolling upgrade of your application with `producerPerConsumerPartition` set to `false` to reduce the number of producers; you should also no longer set the `subBatchPerPartition` container property.
 
 If your brokers are already 2.5 or newer, you should set the `DefaultKafkaProducerFactory` `producerPerConsumerPartition` property to `false`, to reduce the number of producers needed.
@@ -3570,11 +3569,8 @@ When using `V2` mode, it is no longer necessary to set the `subBatchPerPartition
 
 Refer to https://cwiki.apache.org/confluence/display/KAFKA/KIP-447%3A+Producer+scalability+for+exactly+once+semantics[KIP-447] for more information.
 
-IMPORTANT: `kafka-clients` 3.0.0 and later no longer support `V2` (and automatic fallback to `V1`) with brokers earlier than 2.5; you must therefore override the default (`V2`) with `V1` if your brokers are older (or upgrade your brokers).
-
 `V1` and `V2` were previously `ALPHA` and `BETA`; they have been changed to align the framework with https://cwiki.apache.org/confluence/display/KAFKA/KIP-732%3A+Deprecate+eos-alpha+and+replace+eos-beta+with+eos-v2[KIP-732].
 
-
 [[interceptors]]
 ==== Wiring Spring Beans into Producer/Consumer Interceptors
 

spring-kafka-docs/src/main/asciidoc/quick-tour.adoc

@@ -51,7 +51,7 @@ However, the quickest way to get started is to use https://start.spring.io[start
 
 This quick tour works with the following versions:
 
-* Apache Kafka Clients 2.8.0
+* Apache Kafka Clients 3.0.0
 * Spring Framework 5.3.x
 * Minimum Java version: 8
 

spring-kafka-docs/src/main/asciidoc/whats-new.adoc

@@ -6,7 +6,11 @@ For changes in earlier version, see <<history>>.
 [[x28-kafka-client]]
 ==== Kafka Client Version
 
-This version requires the 2.8.0 `kafka-clients
+This version requires the 3.0.0 `kafka-clients
+
+IMPORTANT: When using transactions, `kafka-clients` 3.0.0 and later no longer support `EOSMode.V2` (aka `BETA`) (and automatic fallback to `V1` - aka `ALPHA`) with brokers earlier than 2.5; you must therefore override the default `EOSMode` (`V2`) with `V1` if your brokers are older (or upgrade your brokers).
+
+See <<exactly-once>> and https://cwiki.apache.org/confluence/display/KAFKA/KIP-447%3A+Producer+scalability+for+exactly+once+semantics[KIP-447] for more information.
 
 [[x28-packages]]
 ==== Package Changes

spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java

@@ -55,6 +55,7 @@
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.common.utils.Utils;
 import org.apache.kafka.metadata.BrokerState;
+import org.apache.zookeeper.client.ZKClientConfig;
 import org.apache.zookeeper.server.NIOServerCnxnFactory;
 import org.apache.zookeeper.server.ZooKeeperServer;
 
@@ -67,6 +68,7 @@
 import org.springframework.retry.support.RetryTemplate;
 import org.springframework.util.Assert;
 
+import kafka.cluster.EndPoint;
 import kafka.common.KafkaException;
 import kafka.server.KafkaConfig;
 import kafka.server.KafkaServer;
@@ -557,7 +559,7 @@ public void destroy() {
 	}
 
 	private boolean brokerRunning(KafkaServer kafkaServer) {
-		return !kafkaServer.brokerState().get().equals(BrokerState.NOT_RUNNING);
+		return !kafkaServer.brokerState().equals(BrokerState.NOT_RUNNING);
 	}
 
 	public Set<String> getTopics() {
@@ -584,7 +586,7 @@ public EmbeddedZookeeper getZookeeper() {
 	public synchronized ZooKeeperClient getZooKeeperClient() {
 		if (this.zooKeeperClient == null) {
 			this.zooKeeperClient = new ZooKeeperClient(this.zkConnect, zkSessionTimeout, zkConnectionTimeout,
-					1, Time.SYSTEM, "embeddedKafkaZK", "embeddedKafkaZK");
+					1, Time.SYSTEM, "embeddedKafkaZK", "embeddedKafkaZK", new ZKClientConfig(), "embeddedKafkaZK");
 		}
 		return this.zooKeeperClient;
 	}
@@ -595,7 +597,7 @@ public String getZookeeperConnectionString() {
 
 	public BrokerAddress getBrokerAddress(int i) {
 		KafkaServer kafkaServer = this.kafkaServers.get(i);
-		return new BrokerAddress(LOOPBACK, kafkaServer.config().port());
+		return new BrokerAddress(LOOPBACK, kafkaServer.config().listeners().apply(0).port());
 	}
 
 	public BrokerAddress[] getBrokerAddresses() {
@@ -612,7 +614,8 @@ public int getPartitionsPerTopic() {
 
 	public void bounce(BrokerAddress brokerAddress) {
 		for (KafkaServer kafkaServer : getKafkaServers()) {
-			if (brokerAddress.equals(new BrokerAddress(kafkaServer.config().hostName(), kafkaServer.config().port()))) {
+			EndPoint endpoint = kafkaServer.config().listeners().apply(0);
+			if (brokerAddress.equals(new BrokerAddress(endpoint.host(), endpoint.port()))) {
 				kafkaServer.shutdown();
 				kafkaServer.awaitShutdown();
 			}

spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaStreamsDefaultConfiguration.java

@@ -59,17 +59,14 @@ public class KafkaStreamsDefaultConfiguration {
 	/**
 	 * Bean for the default {@link StreamsBuilderFactoryBean}.
 	 * @param streamsConfigProvider the streams config.
-	 * @param customizerProvider the customizer.
 	 * @param configurerProvider the configurer.
 	 *
 	 * @return the factory bean.
 	 */
-	@SuppressWarnings("deprecation")
 	@Bean(name = DEFAULT_STREAMS_BUILDER_BEAN_NAME)
 	public StreamsBuilderFactoryBean defaultKafkaStreamsBuilder(
 			@Qualifier(DEFAULT_STREAMS_CONFIG_BEAN_NAME)
 					ObjectProvider<KafkaStreamsConfiguration> streamsConfigProvider,
-			ObjectProvider<org.springframework.kafka.config.StreamsBuilderFactoryBeanCustomizer> customizerProvider,
 			ObjectProvider<StreamsBuilderFactoryBeanConfigurer> configurerProvider) {
 
 		KafkaStreamsConfiguration streamsConfig = streamsConfigProvider.getIfAvailable();
@@ -80,11 +77,6 @@ public StreamsBuilderFactoryBean defaultKafkaStreamsBuilder(
 				configurer.configure(fb);
 				configuredBy.add(configurer);
 			});
-			org.springframework.kafka.config.StreamsBuilderFactoryBeanCustomizer customizer = customizerProvider
-					.getIfUnique();
-			if (customizer != null && !configuredBy.contains(customizer)) {
-				customizer.configure(fb);
-			}
 			return fb;
 		}
 		else {

spring-kafka/src/main/java/org/springframework/kafka/config/StreamsBuilderFactoryBeanConfigurer.java

@@ -28,13 +28,12 @@
  *
  */
 @FunctionalInterface
-@SuppressWarnings("deprecation")
-public interface StreamsBuilderFactoryBeanConfigurer extends StreamsBuilderFactoryBeanCustomizer, Ordered {
+public interface StreamsBuilderFactoryBeanConfigurer extends Ordered {
 
 	/**
-	 * Overridden to avoid deprecation warnings.
+	 * Configure the factory bean.
+	 * @param factoryBean the factory bean.
 	 */
-	@Override
 	void configure(StreamsBuilderFactoryBean factoryBean);
 
 	@Override

spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java

@@ -212,7 +212,6 @@ public DefaultKafkaProducerFactory(Map<String, Object> configs,
 			setTransactionIdPrefix(txId);
 			this.configs.remove(ProducerConfig.TRANSACTIONAL_ID_CONFIG);
 		}
-		this.configs.put("internal.auto.downgrade.txn.commit", true);
 	}
 
 	private Supplier<Serializer<K>> keySerializerSupplier(

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaOperations.java

@@ -208,7 +208,9 @@ public interface KafkaOperations<K, V> {
 	 * the container will take care of sending the offsets to the transaction.
 	 * @param offsets The offsets.
 	 * @since 1.3
+	 * @deprecated in the 3.0.0 KafkaProducer.
 	 */
+	@Deprecated
 	void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets);
 
 	/**
@@ -220,7 +222,9 @@ public interface KafkaOperations<K, V> {
 	 * @param offsets The offsets.
 	 * @param consumerGroupId the consumer's group.id.
 	 * @since 1.3
+	 * @deprecated in the 3.0.0 KafkaProducer.
 	 */
+	@Deprecated
 	void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets, String consumerGroupId);
 
 	/**

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java

@@ -546,11 +546,14 @@ public void flush() {
 
 
 	@Override
+	@Deprecated
 	public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets) {
 		sendOffsetsToTransaction(offsets, KafkaUtils.getConsumerGroupId());
 	}
 
+	@SuppressWarnings("deprecation")
 	@Override
+	@Deprecated
 	public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets, String consumerGroupId) {
 		producerForOffsets().sendOffsetsToTransaction(offsets, consumerGroupId);
 	}

spring-kafka/src/main/java/org/springframework/kafka/listener/DefaultAfterRollbackProcessor.java

@@ -128,7 +128,7 @@ private void checkConfig() {
 				"A KafkaOperations is required when 'commitRecovered' is true");
 	}
 
-	@SuppressWarnings({ "unchecked", "rawtypes" })
+	@SuppressWarnings({ "unchecked", "rawtypes", "deprecation" })
 	@Override
 	public void process(List<ConsumerRecord<K, V>> records, Consumer<K, V> consumer,
 			@Nullable MessageListenerContainer container, Exception exception, boolean recoverable, EOSMode eosMode) {

spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

@@ -2689,6 +2689,7 @@ private void sendOffsetsToTransaction() {
 			doSendOffsets(this.producer, commits);
 		}
 
+		@SuppressWarnings("deprecation")
 		private void doSendOffsets(Producer<?, ?> prod, Map<TopicPartition, OffsetAndMetadata> commits) {
 			if (this.eosMode.getMode().equals(EOSMode.V1)) {
 				prod.sendOffsetsToTransaction(commits, this.consumerGroupId);

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/AdapterUtils.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020 the original author or authors.
+ * Copyright 2020-2021 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.
@@ -72,7 +72,7 @@ public static ConsumerRecordMetadata buildConsumerRecordMetadata(Object data) {
 		}
 		ConsumerRecord<?, ?> record = (ConsumerRecord<?, ?>) data;
 		return new ConsumerRecordMetadata(new RecordMetadata(new TopicPartition(record.topic(), record.partition()),
-				0, record.offset(), record.timestamp(), null, record.serializedKeySize(),
+				record.offset(), 0, record.timestamp(), record.serializedKeySize(),
 				record.serializedValueSize()), record.timestampType());
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/streams/messaging/MessagingTransformer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2019-2021 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.
@@ -18,6 +18,7 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Optional;
 
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.producer.ProducerRecord;
@@ -77,9 +78,9 @@ public KeyValue<K, R> transform(K key, V value) {
 		ConsumerRecord<Object, Object> record = new ConsumerRecord<Object, Object>(this.processorContext.topic(),
 				this.processorContext.partition(), this.processorContext.offset(),
 				this.processorContext.timestamp(), TimestampType.NO_TIMESTAMP_TYPE,
-				null, 0, 0,
+				0, 0,
 				key, value,
-				headers);
+				headers, Optional.empty());
 		Message<?> message = this.converter.toMessage(record, null, null, null);
 		message = this.function.exchange(message);
 		List<String> headerList = new ArrayList<>();

spring-kafka/src/test/java/org/springframework/kafka/config/RecordMessagingMessageListenerAdapterTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020 the original author or authors.
+ * Copyright 2020-2021 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.
@@ -21,8 +21,10 @@
 import java.lang.reflect.Method;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Optional;
 
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.header.internals.RecordHeaders;
 import org.apache.kafka.common.record.TimestampType;
 import org.junit.jupiter.api.Test;
 
@@ -51,7 +53,7 @@ void testMetaString() throws Exception {
 		RecordMessagingMessageListenerAdapter<Object, Object> adapter =
 				(RecordMessagingMessageListenerAdapter<Object, Object>) endpoint.createMessageListener(null, null);
 		ConsumerRecord<Object, Object> record = new ConsumerRecord<>("topic", 0, 42L, 43L, TimestampType.CREATE_TIME,
-				0L, 44, 45, null, "foo");
+				44, 45, null, "foo", new RecordHeaders(), Optional.empty());
 		adapter.onMessage(record, null, null);
 		assertThat(bean.string).isEqualTo("foo");
 		assertThat(bean.metadata.topic()).isEqualTo(record.topic());
@@ -86,7 +88,7 @@ else if (method.getName().equals("listen")) {
 		adapter.onMessage(new ConsumerRecord<>("topic", 0, 0L, null, 42), null, null);
 		assertThat(bean.value).isEqualTo(42);
 		ConsumerRecord<Object, Object> record = new ConsumerRecord<>("topic", 0, 42L, 43L, TimestampType.CREATE_TIME,
-				0L, 44, 45, null, "foo");
+				44, 45, null, "foo", new RecordHeaders(), Optional.empty());
 		adapter.onMessage(record, null, null);
 		assertThat(bean.string).isEqualTo("foo");
 		assertThat(bean.metadata.topic()).isEqualTo(record.topic());

spring-kafka/src/test/java/org/springframework/kafka/core/DefaultKafkaProducerFactoryTests.java

@@ -457,23 +457,23 @@ void configUpdates() {
 		Map<String, Object> configs = new HashMap<>();
 		configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
 		DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(configs);
-		assertThat(pf.getConfigurationProperties()).hasSize(2);
+		assertThat(pf.getConfigurationProperties()).hasSize(1);
 		configs.remove(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG);
 		configs.put(ProducerConfig.ACKS_CONFIG, "all");
 		pf.updateConfigs(configs);
-		assertThat(pf.getConfigurationProperties()).hasSize(3);
-		pf.removeConfig(ProducerConfig.ACKS_CONFIG);
 		assertThat(pf.getConfigurationProperties()).hasSize(2);
+		pf.removeConfig(ProducerConfig.ACKS_CONFIG);
+		assertThat(pf.getConfigurationProperties()).hasSize(1);
 		configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "tx-");
 		assertThatIllegalArgumentException().isThrownBy(() -> pf.updateConfigs(configs));
 		configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
 		configs.put(ProducerConfig.CLIENT_ID_CONFIG, "clientId");
 		DefaultKafkaProducerFactory pf1 = new DefaultKafkaProducerFactory<>(configs);
-		assertThat(pf1.getConfigurationProperties()).hasSize(5);
+		assertThat(pf1.getConfigurationProperties()).hasSize(4);
 		configs.put(ProducerConfig.CLIENT_ID_CONFIG, "clientId2");
 		configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "tx2-");
 		pf1.updateConfigs(configs);
-		assertThat(pf1.getConfigurationProperties()).hasSize(5);
+		assertThat(pf1.getConfigurationProperties()).hasSize(4);
 		assertThat(KafkaTestUtils.getPropertyValue(pf1, "clientIdPrefix")).isEqualTo("clientId2");
 		assertThat(KafkaTestUtils.getPropertyValue(pf1, "transactionIdPrefix")).isEqualTo("tx2-");
 		configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, null);

spring-kafka/src/test/java/org/springframework/kafka/core/KafkaAdminTests.java

@@ -96,11 +96,15 @@ public void testAddTopicsAndAddPartitions() throws Exception {
 		int n = 0;
 		await().until(() -> {
 			results.putAll(this.admin.describeTopics("foo", "bar"));
+			TopicDescription foo = results.values().stream()
+					.filter(tp -> tp.name().equals("foo"))
+					.findFirst()
+					.get();
 			TopicDescription bar = results.values().stream()
 					.filter(tp -> tp.name().equals("bar"))
 					.findFirst()
 					.get();
-			return bar.partitions().size() != 1;
+			return foo.partitions().size() == 4 && bar.partitions().size() == 3;
 		});
 		results.forEach((name, td) -> assertThat(td.partitions()).hasSize(name.equals("foo") ? 4 : 3));
 		new DirectFieldAccessor(this.topic1).setPropertyValue("numPartitions", Optional.of(5));

spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java

@@ -330,7 +330,7 @@ public void onError(ProducerRecord<Integer, String> producerRecord, RecordMetada
 		//Drain the topic
 		KafkaTestUtils.getSingleRecord(consumer, INT_KEY_TOPIC);
 		pf.destroy();
-		cpl.onError(records.get(0), new RecordMetadata(new TopicPartition(INT_KEY_TOPIC, -1), 0L, 0L, 0L, 0L, 0, 0),
+		cpl.onError(records.get(0), new RecordMetadata(new TopicPartition(INT_KEY_TOPIC, -1), 0L, 0, 0L, 0, 0),
 				new RuntimeException("x"));
 		assertThat(onErrorDelegateCalls.get()).isEqualTo(2);
 	}

spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTransactionTests.java

@@ -98,6 +98,7 @@ public class KafkaTemplateTransactionTests {
 
 	private final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker();
 
+	@SuppressWarnings("deprecation")
 	@Test
 	public void testLocalTransaction() {
 		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);