Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-1982: Boot 2.6 Compatibility (Retryable Topics)

Browse Source 
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.
This commit is contained in:
Gary Russell
2021-10-29 09:58:20 -04:00
committed by GitHub
parent d2bf7467e4
commit 42fef6e9db
15 changed files with 269 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
spring-kafka-docs/src/main/asciidoc/retrytopic.adoc
View File

@@ -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<Integer, MyPojo> template) {
return RetryTopicConfigurationBuilder
.newInstance()
.dltProcessor(MyCustomDltProcessor.class, "processDltMessage")
.dltProcessor("myCustomDltProcessor", "processDltMessage")
.create(template);
}

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

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

11
spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpointRegistry.java
View File

@@ -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) {

1
spring-kafka/src/main/java/org/springframework/kafka/retrytopic/EndpointCustomizerFactory.java
View File

@@ -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;
/**

87
spring-kafka/src/main/java/org/springframework/kafka/retrytopic/EndpointHandlerMethod.java
View File

@@ -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;
}
}

1
spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfiguration.java
View File

@@ -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.

21
spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationBuilder.java
View File

@@ -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) {

5
spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfigurer.java
View File

@@ -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) {

134
spring-kafka/src/main/java/org/springframework/kafka/support/EndpointHandlerMethod.java Normal file
View File

@@ -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;
}
}

87
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/CircularDltHandlerTests.java Normal file
View File

@@ -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<Object, Object> 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) {
}
}
}

1
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationIntegrationTests.java
View File

@@ -119,6 +119,7 @@ class RetryTopicConfigurationIntegrationTests {
return new KafkaAdmin(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, broker.getBrokersAsString()));
}
@SuppressWarnings("deprecation")
@Bean
RetryTopicConfiguration retryTopicConfiguration1(KafkaTemplate<Integer, String> template) {
return RetryTopicConfigurationBuilder.newInstance()

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

@@ -134,6 +134,7 @@ class RetryTopicConfigurationManualAssignmentIntegrationTests {
return new KafkaAdmin(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, broker.getBrokersAsString()));
}
@SuppressWarnings("deprecation")
@Bean
RetryTopicConfiguration retryTopicConfiguration1(KafkaTemplate<Integer, String> template) {
return RetryTopicConfigurationBuilder.newInstance()

1
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurerTests.java
View File

@@ -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;
/**

2
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicIntegrationTests.java
View File

@@ -314,6 +314,7 @@ public class RetryTopicIntegrationTests {
private static final String DLT_METHOD_NAME = "processDltMessage";
@SuppressWarnings("deprecation")
@Bean
public RetryTopicConfiguration firstRetryTopic(KafkaTemplate<String, String> template) {
return RetryTopicConfigurationBuilder
@@ -327,6 +328,7 @@ public class RetryTopicIntegrationTests {
.create(template);
}
@SuppressWarnings("deprecation")
@Bean
public RetryTopicConfiguration secondRetryTopic(KafkaTemplate<String, String> template) {
return RetryTopicConfigurationBuilder

4
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryableTopicAnnotationProcessorTests.java
View File

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