Upgrade kafka-clients to 3.2.0; other upgrades

Fix deprecation warnings.

Author: garyrussell

build.gradle

@@ -59,16 +59,16 @@ 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.9.0-RC1'
+	micrometerVersion = '1.9.0'
 	mockitoVersion = '4.5.0'
 	reactorVersion = '2020.0.18'
 	scalaVersion = '2.13'
 	springBootVersion = '2.6.7' // docs module
-	springDataVersion = '2021.2.0-RC1'
+	springDataVersion = '2021.2.0'
 	springRetryVersion = '1.3.3'
-	springVersion = '5.3.19'
+	springVersion = '5.3.20'
 	zookeeperVersion = '3.6.3'
 
 	idPrefix = 'kafka'
@@ -155,7 +155,7 @@ subprojects { subproject ->
 	eclipse.project.natures += 'org.springframework.ide.eclipse.core.springnature'
 
 	jacoco {
-		toolVersion = '0.8.7'
+		toolVersion = '0.8.6'
 	}
 
 	// dependencies that are common across all java projects

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/core/KafkaAdmin.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2021 the original author or authors.
+ * Copyright 2017-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.
@@ -209,7 +209,7 @@ public Map<String, TopicDescription> describeTopics(String... topicNames) {
 			Map<String, TopicDescription> results = new HashMap<>();
 			DescribeTopicsResult topics = admin.describeTopics(Arrays.asList(topicNames));
 			try {
-				results.putAll(topics.all().get(this.operationTimeout, TimeUnit.SECONDS));
+				results.putAll(topics.allTopicNames().get(this.operationTimeout, TimeUnit.SECONDS));
 				return results;
 			}
 			catch (InterruptedException ie) {
@@ -251,7 +251,7 @@ private Map<String, NewPartitions> checkPartitions(Map<String, NewTopic> topicNa
 			DescribeTopicsResult topicInfo, List<NewTopic> topicsToAdd) {
 
 		Map<String, NewPartitions> topicsToModify = new HashMap<>();
-		topicInfo.values().forEach((n, f) -> {
+		topicInfo.topicNameValues().forEach((n, f) -> {
 			NewTopic topic = topicNameToTopic.get(n);
 			try {
 				TopicDescription topicDescription = f.get(this.operationTimeout, TimeUnit.SECONDS);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2021 the original author or authors.
+ * Copyright 2016-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.
@@ -480,7 +480,7 @@ protected void checkTopics() {
 								.toArray(String[]::new);
 					}
 					DescribeTopicsResult result = client.describeTopics(Arrays.asList(topics));
-					missing = result.values()
+					missing = result.topicNameValues()
 							.entrySet()
 							.stream()
 							.filter(entry -> {

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.
@@ -62,6 +62,7 @@ class ErrorHandlerAdapter implements CommonErrorHandler {
 		this.batchErrorHandler = batchErrorHandler;
 	}
 
+	@SuppressWarnings("deprecation")
 	@Override
 	public boolean remainingRecords() {
 		return this.errorHandler instanceof RemainingRecordsErrorHandler;
@@ -121,6 +122,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/KafkaAdminTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2021 the original author or authors.
+ * Copyright 2017-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.
@@ -131,7 +131,7 @@ public void testDefaultPartsAndReplicas() throws Exception {
 			await().until(() -> {
 				DescribeTopicsResult topics = adminClient.describeTopics(Arrays.asList("optBoth", "optPart", "optRepl"));
 				try {
-					results.putAll(topics.all().get(10, TimeUnit.SECONDS));
+					results.putAll(topics.allTopicNames().get(10, TimeUnit.SECONDS));
 					return true;
 				}
 				catch (InterruptedException ie) {

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);

spring-kafka/src/test/java/org/springframework/kafka/listener/ManualNackRecordTests.java

@@ -246,6 +246,7 @@ public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() {
 			factory.getContainerProperties().setMissingTopicsFatal(false);
 			factory.setRecordInterceptor(new RecordInterceptor() {
 
+				@SuppressWarnings("deprecation")
 				@Override
 				@Nullable
 				public ConsumerRecord intercept(ConsumerRecord record) {