Upgrade kafka-clients to 3.2.0; other upgrades

Fix deprecation warnings.

Author: garyrussell

build.gradle

@@ -64,17 +64,17 @@ ext {
 	jaywayJsonPathVersion = '2.6.0'
 	junit4Version = '4.13.2'
 	junitJupiterVersion = '5.8.2'
-	kafkaVersion = '3.1.0'
+	kafkaVersion = '3.2.0'
 	log4jVersion = '2.17.2'
 	micrometerVersion = '1.10.0-SNAPSHOT'
 	micrometerTracingVersion = '1.0.0-SNAPSHOT'
 	mockitoVersion = '4.0.0'
-	reactorVersion = '2020.0.17'
+	reactorVersion = '2020.0.18'
 	scalaVersion = '2.13'
-	springBootVersion = '2.6.4' // docs module
-	springDataVersion = '2022.0.0-M3'
-	springRetryVersion = '1.3.2'
-	springVersion = '6.0.0-M3'
+	springBootVersion = '2.6.7' // docs module
+	springDataVersion = '2022.0.0-M4'
+	springRetryVersion = '1.3.3'
+	springVersion = '6.0.0-M4'
 	zookeeperVersion = '3.6.3'
 
 	idPrefix = 'kafka'

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

@@ -87,7 +87,6 @@
 import org.springframework.kafka.listener.KafkaListenerErrorHandler;
 import org.springframework.kafka.listener.adapter.RecordFilterStrategy;
 import org.springframework.kafka.retrytopic.RetryTopicBeanNames;
-import org.springframework.kafka.retrytopic.RetryTopicBootstrapper;
 import org.springframework.kafka.retrytopic.RetryTopicConfiguration;
 import org.springframework.kafka.retrytopic.RetryTopicConfigurer;
 import org.springframework.kafka.support.TopicPartitionOffset;
@@ -527,8 +526,9 @@ private void bootstrapRetryTopicIfNecessary() {
 		BeanDefinitionRegistry registry = (BeanDefinitionRegistry) this.beanFactory;
 		if (!registry.containsBeanDefinition("internalRetryTopicBootstrapper")) {
 			registry.registerBeanDefinition("internalRetryTopicBootstrapper",
-					new RootBeanDefinition(RetryTopicBootstrapper.class));
-			this.beanFactory.getBean("internalRetryTopicBootstrapper", RetryTopicBootstrapper.class).bootstrapRetryTopic();
+					new RootBeanDefinition(org.springframework.kafka.retrytopic.RetryTopicBootstrapper.class));
+			this.beanFactory.getBean("internalRetryTopicBootstrapper",
+					org.springframework.kafka.retrytopic.RetryTopicBootstrapper.class).bootstrapRetryTopic();
 		}
 	}
 

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

@@ -37,7 +37,6 @@
 import org.springframework.kafka.retrytopic.RetryTopicConfigurationBuilder;
 import org.springframework.kafka.retrytopic.RetryTopicConfigurer;
 import org.springframework.kafka.retrytopic.RetryTopicConstants;
-import org.springframework.kafka.retrytopic.RetryTopicInternalBeanNames;
 import org.springframework.kafka.support.EndpointHandlerMethod;
 import org.springframework.retry.annotation.Backoff;
 import org.springframework.retry.backoff.ExponentialBackOffPolicy;
@@ -215,7 +214,8 @@ private EndpointHandlerMethod getDltProcessor(Method listenerMethod, Object bean
 			}
 		}
 		try {
-			return this.beanFactory.getBean(RetryTopicInternalBeanNames.DEFAULT_KAFKA_TEMPLATE_BEAN_NAME,
+			return this.beanFactory.getBean(
+					org.springframework.kafka.retrytopic.RetryTopicInternalBeanNames.DEFAULT_KAFKA_TEMPLATE_BEAN_NAME,
 					KafkaOperations.class);
 		}
 		catch (NoSuchBeanDefinitionException ex) {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021 the original author or authors.
+ * Copyright 2021-2022 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,6 +63,7 @@ class ErrorHandlerAdapter implements CommonErrorHandler {
 		this.batchErrorHandler = batchErrorHandler;
 	}
 
+	@SuppressWarnings("deprecation")
 	@Override
 	public boolean remainingRecords() {
 		return this.errorHandler instanceof RemainingRecordsErrorHandler;
@@ -122,6 +123,7 @@ public void handleOtherException(Exception thrownException, Consumer<?, ?> consu
 		}
 	}
 
+	@SuppressWarnings("deprecation")
 	@Override
 	public void handleRecord(Exception thrownException, ConsumerRecord<?, ?> record, Consumer<?, ?> consumer,
 			MessageListenerContainer container) {

spring-kafka/src/test/java/org/springframework/kafka/core/reactive/ReactiveKafkaProducerTemplateTransactionIntegrationTests.java

@@ -269,7 +269,7 @@ public void shouldSendOneRecordTransactionallyViaTemplateAsSenderRecordAndReceiv
 				.onErrorResume(error -> reactiveKafkaProducerTemplate.transactionManager()
 						.abort()
 						.then(Mono.error(error))))
-				.expectErrorMatches(throwable -> throwable instanceof KafkaException &&
+				.expectErrorMatches(throwable -> throwable instanceof IllegalStateException &&
 						throwable.getMessage().equals("TransactionalId reactive.transaction: Invalid transition " +
 								"attempted from state READY to state ABORTING_TRANSACTION"))
 				.verify(DEFAULT_VERIFY_TIMEOUT);