Fix deprecation warnings

* Remove and already deprecated `KafkaConsumer.close(Duration)`
 method override from the `DefaultKafkaConsumerFactory`.
Use `close(CloseOptions option)` instead
* Replace a deprecated `org.springframework.context.expression.MapAccessor`
 with a new `org.springframework.expression.spel.support.MapAccessor`
* Fix `DefaultKafkaConsumerFactoryTests` & `RetryTopicConfigurationManualAssignmentIntegrationTests`
to not use a deprecated `KafkaConsumer.close(Duration)`, but based on the `CloseOptions.timeout()` instead

Author: artembilan

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

@@ -16,7 +16,6 @@
 
 package org.springframework.kafka.core;
 
-import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Enumeration;
@@ -32,6 +31,7 @@
 import java.util.function.Supplier;
 
 import org.apache.commons.logging.LogFactory;
+import org.apache.kafka.clients.consumer.CloseOptions;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
@@ -524,8 +524,8 @@ public void close() {
 		}
 
 		@Override
-		public void close(Duration timeout) {
-			super.close(timeout);
+		public void close(CloseOptions option) {
+			super.close(option);
 			notifyConsumerRemoved();
 		}
 

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

@@ -43,13 +43,13 @@
 import org.jspecify.annotations.Nullable;
 import reactor.core.publisher.Mono;
 
-import org.springframework.context.expression.MapAccessor;
 import org.springframework.core.MethodParameter;
 import org.springframework.core.log.LogAccessor;
 import org.springframework.expression.BeanResolver;
 import org.springframework.expression.Expression;
 import org.springframework.expression.common.LiteralExpression;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.expression.spel.support.MapAccessor;
 import org.springframework.expression.spel.support.StandardEvaluationContext;
 import org.springframework.expression.spel.support.StandardTypeConverter;
 import org.springframework.kafka.core.KafkaTemplate;
@@ -735,8 +735,8 @@ protected void asyncFailure(Object request, @Nullable Acknowledgment acknowledgm
 		try {
 			Throwable cause = t instanceof CompletionException ? t.getCause() : t;
 			handleException(request, acknowledgment, consumer, source,
-							new ListenerExecutionFailedException(createMessagingErrorMessage(
-									"Async Fail", Objects.requireNonNull(source).getPayload()), cause));
+					new ListenerExecutionFailedException(createMessagingErrorMessage(
+							"Async Fail", Objects.requireNonNull(source).getPayload()), cause));
 		}
 		catch (Throwable ex) {
 			acknowledge(acknowledgment);

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

@@ -31,6 +31,7 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
+import org.apache.kafka.clients.consumer.CloseOptions;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
@@ -490,7 +491,7 @@ public void consumerRemoved(String id, Consumer consumer) {
 		assertThat(adds.get(0)).isEqualTo("cf.foo-0");
 		assertThat(removals).isEmpty();
 		if (closeWithTimeout) {
-			consumer.close(Duration.ofSeconds(10));
+			consumer.close(CloseOptions.timeout(Duration.ofSeconds(10)));
 		}
 		else {
 			consumer.close();

spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationManualAssignmentIntegrationTests.java

@@ -23,6 +23,7 @@
 import java.util.concurrent.TimeUnit;
 
 import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.consumer.CloseOptions;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.common.PartitionInfo;
@@ -55,6 +56,8 @@
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 2.7.7
  *
  */
@@ -89,7 +92,7 @@ void includeTopic(@Autowired EmbeddedKafkaBroker broker, @Autowired ConsumerFact
 			assertThat(config.latch.await(120, TimeUnit.SECONDS)).isTrue();
 		}
 		finally {
-			consumer.close(Duration.ofSeconds(10));
+			consumer.close(CloseOptions.timeout(Duration.ofSeconds(10)));
 		}
 	}