From 42fef6e9db65e8c409b91c7e231032a30e9b40f3 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 29 Oct 2021 09:58:20 -0400 Subject: [PATCH] GH-1982: Boot 2.6 Compatibility (Retryable Topics) Resolves https://github.com/spring-projects/spring-kafka/issues/1982 Spring Boot 2.6 disables circular bean references by default. When using `RetryTopicConfigurationBuilder.dltHandlerMethod` to specify the location of the bean handler, a circular reference is caused if the method is in the same bean as the listener (because the `EndpointHandlerMethod` attempts to resolve it while the listener bean is being post processed). Deprecate `RetryTopicConfigurationBuilder.dltHandlerMethod(Class clazz, String methodName)`. As an aside, it is antithetical in Spring to expect a single bean with a type. Replace with `dltHandlerMethod(String beanName, String methodName)`. This allows deferring the bean lookup if a `BeanCurrentlyInCreationException` is thrown. When that happens, return the "endpoint" as the "bean" and a dummy `Method` to satisfy nullable requirements. When the registry detects that the "bean" is an "endpoint", resolve the bean and method at that time and re-populate the `MethodKafkaListenerEndpoint`. Move `EndpointHandlerMethod` to `support` to avoid a package tangle. Add a test to simulate Boot 2.6 default behavior. * Remove deprecated method from docs. --- .../src/main/asciidoc/retrytopic.adoc | 5 +- .../RetryableTopicAnnotationProcessor.java | 2 +- .../config/KafkaListenerEndpointRegistry.java | 11 ++ .../retrytopic/EndpointCustomizerFactory.java | 1 + .../retrytopic/EndpointHandlerMethod.java | 87 ------------ .../retrytopic/RetryTopicConfiguration.java | 1 + .../RetryTopicConfigurationBuilder.java | 21 +++ .../retrytopic/RetryTopicConfigurer.java | 5 +- .../kafka/support/EndpointHandlerMethod.java | 134 ++++++++++++++++++ .../retrytopic/CircularDltHandlerTests.java | 87 ++++++++++++ ...tryTopicConfigurationIntegrationTests.java | 1 + ...ationManualAssignmentIntegrationTests.java | 1 + .../retrytopic/RetryTopicConfigurerTests.java | 1 + .../RetryTopicIntegrationTests.java | 2 + ...etryableTopicAnnotationProcessorTests.java | 4 +- 15 files changed, 269 insertions(+), 94 deletions(-) delete mode 100644 spring-kafka/src/main/java/org/springframework/kafka/retrytopic/EndpointHandlerMethod.java create mode 100644 spring-kafka/src/main/java/org/springframework/kafka/support/EndpointHandlerMethod.java create mode 100644 spring-kafka/src/test/java/org/springframework/kafka/retrytopic/CircularDltHandlerTests.java diff --git a/spring-kafka-docs/src/main/asciidoc/retrytopic.adoc b/spring-kafka-docs/src/main/asciidoc/retrytopic.adoc index 6aaf86b5..160fca80 100644 --- a/spring-kafka-docs/src/main/asciidoc/retrytopic.adoc +++ b/spring-kafka-docs/src/main/asciidoc/retrytopic.adoc @@ -500,8 +500,7 @@ public void processMessage(MyPojo message) { ---- ==== -The DLT handler method can also be provided through the RetryTopicConfigurationBuilder.dltHandlerMethod(Class, String) method, passing as arguments the class and method name that should process the DLT's messages. -If a bean instance of the provided class is found in the application context that bean is used for Dlt processing, otherwise an instance is created with full dependency injection support. +The DLT handler method can also be provided through the RetryTopicConfigurationBuilder.dltHandlerMethod(String, String) method, passing as arguments the bean name and method name that should process the DLT's messages. ==== [source, java] @@ -510,7 +509,7 @@ If a bean instance of the provided class is found in the application context tha public RetryTopicConfiguration myRetryTopic(KafkaTemplate template) { return RetryTopicConfigurationBuilder .newInstance() - .dltProcessor(MyCustomDltProcessor.class, "processDltMessage") + .dltProcessor("myCustomDltProcessor", "processDltMessage") .create(template); } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/annotation/RetryableTopicAnnotationProcessor.java b/spring-kafka/src/main/java/org/springframework/kafka/annotation/RetryableTopicAnnotationProcessor.java index e7c5bc77..fc10198f 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/annotation/RetryableTopicAnnotationProcessor.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/annotation/RetryableTopicAnnotationProcessor.java @@ -32,12 +32,12 @@ import org.springframework.context.expression.StandardBeanExpressionResolver; import org.springframework.core.annotation.AnnotationUtils; import org.springframework.expression.spel.support.StandardEvaluationContext; import org.springframework.kafka.core.KafkaOperations; -import org.springframework.kafka.retrytopic.EndpointHandlerMethod; import org.springframework.kafka.retrytopic.RetryTopicConfiguration; 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; import org.springframework.retry.backoff.ExponentialRandomBackOffPolicy; diff --git a/spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpointRegistry.java b/spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpointRegistry.java index 34129e71..09990e82 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpointRegistry.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpointRegistry.java @@ -42,6 +42,7 @@ import org.springframework.kafka.listener.AbstractMessageListenerContainer; import org.springframework.kafka.listener.ContainerGroup; import org.springframework.kafka.listener.ListenerContainerRegistry; import org.springframework.kafka.listener.MessageListenerContainer; +import org.springframework.kafka.support.EndpointHandlerMethod; import org.springframework.lang.Nullable; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -214,6 +215,16 @@ public class KafkaListenerEndpointRegistry implements ListenerContainerRegistry, protected MessageListenerContainer createListenerContainer(KafkaListenerEndpoint endpoint, KafkaListenerContainerFactory factory) { + if (endpoint instanceof MethodKafkaListenerEndpoint) { + MethodKafkaListenerEndpoint mkle = (MethodKafkaListenerEndpoint) endpoint; + Object bean = mkle.getBean(); + if (bean instanceof EndpointHandlerMethod) { + EndpointHandlerMethod ehm = (EndpointHandlerMethod) bean; + ehm = new EndpointHandlerMethod(ehm.resolveBean(this.applicationContext), ehm.getMethodName()); + mkle.setBean(ehm.resolveBean(this.applicationContext)); + mkle.setMethod(ehm.getMethod()); + } + } MessageListenerContainer listenerContainer = factory.createListenerContainer(endpoint); if (listenerContainer instanceof InitializingBean) { diff --git a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/EndpointCustomizerFactory.java b/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/EndpointCustomizerFactory.java index dd7aedf5..1f39fc8c 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/EndpointCustomizerFactory.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/EndpointCustomizerFactory.java @@ -23,6 +23,7 @@ import java.util.stream.Collectors; import org.springframework.beans.factory.BeanFactory; import org.springframework.kafka.config.MethodKafkaListenerEndpoint; +import org.springframework.kafka.support.EndpointHandlerMethod; import org.springframework.kafka.support.TopicPartitionOffset; /** diff --git a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/EndpointHandlerMethod.java b/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/EndpointHandlerMethod.java deleted file mode 100644 index 629f14a2..00000000 --- a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/EndpointHandlerMethod.java +++ /dev/null @@ -1,87 +0,0 @@ -/* - * Copyright 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. - * You may obtain a copy of the License at - * - * https://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.kafka.retrytopic; - -import java.lang.reflect.Method; -import java.util.Arrays; - -import org.springframework.beans.factory.BeanFactory; -import org.springframework.beans.factory.NoSuchBeanDefinitionException; -import org.springframework.beans.factory.support.BeanDefinitionRegistry; -import org.springframework.beans.factory.support.RootBeanDefinition; -import org.springframework.util.Assert; -import org.springframework.util.ReflectionUtils; - -/** - * Handler method for retrying endpoints. - * - * @author Tomaz Fernandes - * @author Gary Russell - * @since 2.7 - * - */ -public class EndpointHandlerMethod { - - private final Class beanClass; - - private final Method method; - - private Object bean; - - public EndpointHandlerMethod(Class beanClass, String methodName) { - Assert.notNull(beanClass, () -> "No destination bean class provided!"); - Assert.notNull(methodName, () -> "No method name for destination bean class provided!"); - this.method = Arrays.stream(ReflectionUtils.getDeclaredMethods(beanClass)) - .filter(mthd -> mthd.getName().equals(methodName)) - .findFirst() - .orElseThrow(() -> new IllegalArgumentException( - String.format("No method %s in class %s", methodName, beanClass))); - this.beanClass = beanClass; - } - - public EndpointHandlerMethod(Object bean, Method method) { - Assert.notNull(bean, () -> "No bean for destination provided!"); - Assert.notNull(method, () -> "No method for destination bean class provided!"); - this.method = method; - this.bean = bean; - this.beanClass = bean.getClass(); - } - - /** - * Return the method. - * @return the method. - */ - public Method getMethod() { - return this.method; - } - - public Object resolveBean(BeanFactory beanFactory) { - if (this.bean == null) { - try { - this.bean = beanFactory.getBean(this.beanClass); - } - catch (NoSuchBeanDefinitionException e) { - String beanName = this.beanClass.getSimpleName() + "-handlerMethod"; - ((BeanDefinitionRegistry) beanFactory).registerBeanDefinition(beanName, - new RootBeanDefinition(this.beanClass)); - this.bean = beanFactory.getBean(beanName); - } - } - return this.bean; - } - -} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfiguration.java b/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfiguration.java index 380078a6..ffde6dc5 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfiguration.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfiguration.java @@ -19,6 +19,7 @@ package org.springframework.kafka.retrytopic; import java.util.List; import org.springframework.kafka.support.AllowDenyCollectionManager; +import org.springframework.kafka.support.EndpointHandlerMethod; /** * Contains the provided configuration for the retryable topics. diff --git a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationBuilder.java b/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationBuilder.java index d2430f23..2ac0680c 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationBuilder.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationBuilder.java @@ -24,6 +24,7 @@ import org.springframework.classify.BinaryExceptionClassifierBuilder; import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; import org.springframework.kafka.core.KafkaOperations; import org.springframework.kafka.support.AllowDenyCollectionManager; +import org.springframework.kafka.support.EndpointHandlerMethod; import org.springframework.lang.Nullable; import org.springframework.retry.backoff.BackOffPolicy; import org.springframework.retry.backoff.ExponentialBackOffPolicy; @@ -80,11 +81,31 @@ public class RetryTopicConfigurationBuilder { private Boolean autoStartDltHandler; /* ---------------- DLT Behavior -------------- */ + /** + * Configure a DLT handler method. + * @param clazz the class containing the method. + * @param methodName the method name. + * @return the builder. + * @deprecated in favor of {@link #dltHandlerMethod(String, String)}. + */ + @Deprecated public RetryTopicConfigurationBuilder dltHandlerMethod(Class clazz, String methodName) { this.dltHandlerMethod = RetryTopicConfigurer.createHandlerMethodWith(clazz, methodName); return this; } + /** + * Configure a DLT handler method. + * @param beanName the bean name. + * @param methodName the method name. + * @return the builder. + * @since 2.8 + */ + public RetryTopicConfigurationBuilder dltHandlerMethod(String beanName, String methodName) { + this.dltHandlerMethod = RetryTopicConfigurer.createHandlerMethodWith(beanName, methodName); + return this; + } + public RetryTopicConfigurationBuilder dltHandlerMethod( EndpointHandlerMethod endpointHandlerMethod) { diff --git a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfigurer.java b/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfigurer.java index bb738ebd..9fba6cf3 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfigurer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfigurer.java @@ -34,6 +34,7 @@ import org.springframework.kafka.config.KafkaListenerEndpointRegistrar; import org.springframework.kafka.config.MethodKafkaListenerEndpoint; import org.springframework.kafka.config.MultiMethodKafkaListenerEndpoint; import org.springframework.kafka.listener.ListenerUtils; +import org.springframework.kafka.support.EndpointHandlerMethod; import org.springframework.lang.Nullable; @@ -387,8 +388,8 @@ public class RetryTopicConfigurer { } } - public static EndpointHandlerMethod createHandlerMethodWith(Class beanClass, String methodName) { - return new EndpointHandlerMethod(beanClass, methodName); + public static EndpointHandlerMethod createHandlerMethodWith(Object beanOrClass, String methodName) { + return new EndpointHandlerMethod(beanOrClass, methodName); } public static EndpointHandlerMethod createHandlerMethodWith(Object bean, Method method) { diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/EndpointHandlerMethod.java b/spring-kafka/src/main/java/org/springframework/kafka/support/EndpointHandlerMethod.java new file mode 100644 index 00000000..9d86bce5 --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/EndpointHandlerMethod.java @@ -0,0 +1,134 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.kafka.support; + +import java.lang.reflect.Method; +import java.util.Arrays; + +import org.springframework.beans.factory.BeanCurrentlyInCreationException; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.NoSuchBeanDefinitionException; +import org.springframework.beans.factory.support.BeanDefinitionRegistry; +import org.springframework.beans.factory.support.RootBeanDefinition; +import org.springframework.util.Assert; +import org.springframework.util.ReflectionUtils; + +/** + * Handler method for retrying endpoints. + * + * @author Tomaz Fernandes + * @author Gary Russell + * @since 2.7 + * + */ +public class EndpointHandlerMethod { + + private final Object beanOrClass; + + private final String methodName; + + private Object bean; + + private Method method; + + public EndpointHandlerMethod(Object beanOrClass, String methodName) { + Assert.notNull(beanOrClass, () -> "No destination bean or class provided!"); + Assert.notNull(methodName, () -> "No method name for destination bean class provided!"); + this.beanOrClass = beanOrClass; + this.methodName = methodName; + } + + public EndpointHandlerMethod(Object bean, Method method) { + Assert.notNull(bean, () -> "No bean for destination provided!"); + Assert.notNull(method, () -> "No method for destination bean class provided!"); + this.method = method; + this.bean = bean; + this.beanOrClass = bean.getClass(); + this.methodName = method.getName(); + } + + /** + * Return the method. + * @return the method. + */ + public Method getMethod() { + if (this.beanOrClass instanceof Class) { + return forClass((Class) this.beanOrClass); + } + Assert.state(this.bean != null, "Bean must be resolved before accessing its method"); + if (this.bean instanceof EndpointHandlerMethod) { + try { + return Object.class.getMethod("toString"); + } + catch (NoSuchMethodException | SecurityException e) { + } + } + return forClass(this.bean.getClass()); + } + + /** + * Return the method name. + * @return the name. + * @since 2.8 + */ + public String getMethodName() { + Assert.state(this.methodName != null, "Unexpected call to getMethodName()"); + return this.methodName; + } + + public Object resolveBean(BeanFactory beanFactory) { + if (this.bean instanceof EndpointHandlerMethod) { + return ((EndpointHandlerMethod) this.bean).beanOrClass; + } + if (this.bean == null) { + try { + if (this.beanOrClass instanceof Class) { + Class clazz = (Class) this.beanOrClass; + try { + this.bean = beanFactory.getBean(clazz); + } + catch (NoSuchBeanDefinitionException e) { + String beanName = clazz.getSimpleName() + "-handlerMethod"; + ((BeanDefinitionRegistry) beanFactory).registerBeanDefinition(beanName, + new RootBeanDefinition(clazz)); + this.bean = beanFactory.getBean(beanName); + } + } + else { + String beanName = (String) this.beanOrClass; + this.bean = beanFactory.getBean(beanName); + } + } + catch (BeanCurrentlyInCreationException ex) { + this.bean = this; + } + } + return this.bean; + } + + private Method forClass(Class clazz) { + if (this.method == null) { + this.method = Arrays.stream(ReflectionUtils.getDeclaredMethods(clazz)) + .filter(mthd -> mthd.getName().equals(this.methodName)) + .findFirst() + .orElseThrow(() -> new IllegalArgumentException( + String.format("No method %s in class %s", this.methodName, clazz))); + } + return this.method; + } + +} diff --git a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/CircularDltHandlerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/CircularDltHandlerTests.java new file mode 100644 index 00000000..cd9ce880 --- /dev/null +++ b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/CircularDltHandlerTests.java @@ -0,0 +1,87 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.kafka.retrytopic; + +import static org.mockito.Mockito.mock; + +import org.junit.jupiter.api.Test; + +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.kafka.annotation.EnableKafka; +import org.springframework.kafka.annotation.KafkaListener; +import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; +import org.springframework.kafka.core.ConsumerFactory; +import org.springframework.kafka.core.KafkaTemplate; + +/** + * @author Gary Russell + * @since 2.8 + * + */ +public class CircularDltHandlerTests { + + @Test + void contextLoads() { + AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(); + context.register(Config.class); + context.setAllowCircularReferences(false); + context.refresh(); + } + + @Configuration + @EnableKafka + public static class Config { + + @SuppressWarnings("unchecked") + @Bean + ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() { + ConcurrentKafkaListenerContainerFactory factory = + new ConcurrentKafkaListenerContainerFactory<>(); + factory.setConsumerFactory(mock(ConsumerFactory.class)); + return factory; + } + + @Bean + RetryTopicConfiguration retryConfig() { + return RetryTopicConfigurationBuilder + .newInstance() + .maxAttempts(1) + .dltHandlerMethod("listener", "dlt") + .create(mock(KafkaTemplate.class)); + } + + @Bean + Listener listener() { + return new Listener(); + } + + } + + public static class Listener { + + @KafkaListener(id = "test", topics = "test", autoStartup = "false") + void listen(String in) { + } + + public void dlt(String in) { + } + + } + +} diff --git a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationIntegrationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationIntegrationTests.java index 5b1acb5d..5cb13a7d 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationIntegrationTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationIntegrationTests.java @@ -119,6 +119,7 @@ class RetryTopicConfigurationIntegrationTests { return new KafkaAdmin(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, broker.getBrokersAsString())); } + @SuppressWarnings("deprecation") @Bean RetryTopicConfiguration retryTopicConfiguration1(KafkaTemplate template) { return RetryTopicConfigurationBuilder.newInstance() diff --git a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationManualAssignmentIntegrationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationManualAssignmentIntegrationTests.java index 4ab219ea..267298f4 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationManualAssignmentIntegrationTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationManualAssignmentIntegrationTests.java @@ -134,6 +134,7 @@ class RetryTopicConfigurationManualAssignmentIntegrationTests { return new KafkaAdmin(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, broker.getBrokersAsString())); } + @SuppressWarnings("deprecation") @Bean RetryTopicConfiguration retryTopicConfiguration1(KafkaTemplate template) { return RetryTopicConfigurationBuilder.newInstance() diff --git a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurerTests.java index 9eac8e09..e38911d8 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurerTests.java @@ -52,6 +52,7 @@ import org.springframework.kafka.config.KafkaListenerContainerFactory; import org.springframework.kafka.config.KafkaListenerEndpointRegistrar; import org.springframework.kafka.config.MethodKafkaListenerEndpoint; import org.springframework.kafka.config.MultiMethodKafkaListenerEndpoint; +import org.springframework.kafka.support.EndpointHandlerMethod; import org.springframework.test.util.ReflectionTestUtils; /** diff --git a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicIntegrationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicIntegrationTests.java index 265e7225..5a79d668 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicIntegrationTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicIntegrationTests.java @@ -314,6 +314,7 @@ public class RetryTopicIntegrationTests { private static final String DLT_METHOD_NAME = "processDltMessage"; + @SuppressWarnings("deprecation") @Bean public RetryTopicConfiguration firstRetryTopic(KafkaTemplate template) { return RetryTopicConfigurationBuilder @@ -327,6 +328,7 @@ public class RetryTopicIntegrationTests { .create(template); } + @SuppressWarnings("deprecation") @Bean public RetryTopicConfiguration secondRetryTopic(KafkaTemplate template) { return RetryTopicConfigurationBuilder diff --git a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryableTopicAnnotationProcessorTests.java b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryableTopicAnnotationProcessorTests.java index 57f53fbb..efa86471 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryableTopicAnnotationProcessorTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryableTopicAnnotationProcessorTests.java @@ -40,6 +40,7 @@ import org.springframework.kafka.annotation.KafkaListener; import org.springframework.kafka.annotation.RetryableTopic; import org.springframework.kafka.annotation.RetryableTopicAnnotationProcessor; import org.springframework.kafka.core.KafkaOperations; +import org.springframework.kafka.support.EndpointHandlerMethod; import org.springframework.retry.annotation.Backoff; import org.springframework.test.util.ReflectionTestUtils; import org.springframework.util.ReflectionUtils; @@ -136,7 +137,8 @@ class RetryableTopicAnnotationProcessorTests { // then EndpointHandlerMethod dltHandlerMethod = configuration.getDltHandlerMethod(); - Method method = (Method) ReflectionTestUtils.getField(dltHandlerMethod, "method"); + dltHandlerMethod.resolveBean(this.beanFactory); + Method method = dltHandlerMethod.getMethod(); assertThat(method.getName()) .isEqualTo(RetryTopicConfigurer.LoggingDltListenerHandlerMethod.DEFAULT_DLT_METHOD_NAME);