Package org.springframework.kafka.retrytopic

Class RetryTopicConfigurationSupport

java.lang.Object

org.springframework.kafka.retrytopic.RetryTopicConfigurationSupport

All Implemented Interfaces:

Aware, SmartInitializingSingleton, ApplicationContextAware

public class RetryTopicConfigurationSupport
extends Object
implements ApplicationContextAware, SmartInitializingSingleton

This is the main class providing the configuration behind the non-blocking, topic-based delayed retries feature. It is typically imported by adding @EnableKafkaRetryTopic to an application @Configuration class. An alternative more advanced option is to extend directly from this class and override methods as necessary, remembering to add @Configuration to the subclass and @Bean to overridden @Bean methods. For more details see the javadoc of @EnableRetryTopic.

Since:

2.9

Author:

Tomaz Fernandes, Gary Russell

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static class	RetryTopicConfigurationSupport.BlockingRetriesConfigurer	Configure blocking retries to be used along non-blocking.
static class	RetryTopicConfigurationSupport.CustomizersConfigurer	Configure customizers for components instantiated by the retry topics feature.

Constructor Summary

Constructors

Constructor	Description
RetryTopicConfigurationSupport()

Method Summary

Modifier and Type	Method	Description
void	afterSingletonsInstantiated()
protected void	configureBlockingRetries(RetryTopicConfigurationSupport.BlockingRetriesConfigurer blockingRetries)	Override this method to configure blocking retries parameters such as exceptions to be retried and the BackOff to be used.
protected void	configureCustomizers(RetryTopicConfigurationSupport.CustomizersConfigurer customizersConfigurer)	Override this method to configure customizers for components created by non-blocking retries' configuration, such as MessageListenerContainer, DeadLetterPublishingRecoverer and DefaultErrorHandler.
protected Consumer<DeadLetterPublishingRecovererFactory>	configureDeadLetterPublishingContainerFactory()	Override this method to further configure the DeadLetterPublishingRecovererFactory.
protected Consumer<DestinationTopicResolver>	configureDestinationTopicResolver()	Override this method to configure the DestinationTopicResolver.
protected Consumer<ListenerContainerFactoryConfigurer>	configureListenerContainerFactoryConfigurer()	Override this method to further configure the ListenerContainerFactoryConfigurer.
protected Consumer<RetryTopicConfigurer>	configureRetryTopicConfigurer()	Override this method if you need to configure the RetryTopicConfigurer.
protected RetryTopicComponentFactory	createComponentFactory()	Override this method to provide a subclass of RetryTopicComponentFactory with different component implementations or subclasses.
DestinationTopicResolver	destinationTopicResolver(ObjectProvider<RetryTopicComponentFactory> componentFactoryProvider)	Return a global DestinationTopicResolver for resolving the DestinationTopic to which a given ConsumerRecord should be sent for retry.
KafkaConsumerBackoffManager	kafkaConsumerBackoffManager(ApplicationContext applicationContext, ListenerContainerRegistry registry, ObjectProvider<RetryTopicComponentFactory> componentFactoryProvider, RetryTopicSchedulerWrapper wrapper, TaskScheduler taskScheduler)	Create the KafkaConsumerBackoffManager instance that will be used to back off partitions.
protected void	manageNonBlockingFatalExceptions(List<Class<? extends Throwable>> nonBlockingRetriesExceptions)	Override this method to manage non-blocking retries fatal exceptions.
RetryTopicConfigurer	retryTopicConfigurer(KafkaConsumerBackoffManager kafkaConsumerBackoffManager, DestinationTopicResolver destinationTopicResolver, ObjectProvider<RetryTopicComponentFactory> componentFactoryProvider, BeanFactory beanFactory)	Return a global RetryTopicConfigurer for configuring retry topics for KafkaListenerEndpoint instances with a corresponding RetryTopicConfiguration.
void	setApplicationContext(ApplicationContext applicationContext)

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Details

RetryTopicConfigurationSupport

public RetryTopicConfigurationSupport()

Method Details

setApplicationContext

public void setApplicationContext(ApplicationContext applicationContext)
                           throws BeansException

Specified by:

setApplicationContext in interface ApplicationContextAware

Throws:

BeansException

afterSingletonsInstantiated

public void afterSingletonsInstantiated()

Specified by:

afterSingletonsInstantiated in interface SmartInitializingSingleton

retryTopicConfigurer

@Bean(name="org.springframework.kafka.retrytopic.internalRetryTopicConfigurer")
public RetryTopicConfigurer retryTopicConfigurer(@Qualifier("org.springframework.kafka.config.internalKafkaConsumerBackOffManager")
 KafkaConsumerBackoffManager kafkaConsumerBackoffManager,
 @Qualifier("org.springframework.kafka.retrytopic.internalDestinationTopicResolver")
 DestinationTopicResolver destinationTopicResolver,
 ObjectProvider<RetryTopicComponentFactory> componentFactoryProvider,
 BeanFactory beanFactory)

Return a global RetryTopicConfigurer for configuring retry topics for KafkaListenerEndpoint instances with a corresponding RetryTopicConfiguration. To configure it, consider overriding the configureRetryTopicConfigurer().

Parameters:

kafkaConsumerBackoffManager - the global KafkaConsumerBackoffManager.

destinationTopicResolver - the global DestinationTopicResolver.

componentFactoryProvider - the component factory provider.

beanFactory - the BeanFactory.

Returns:

the instance.

See Also:

• KafkaListenerAnnotationBeanPostProcessor

configureRetryTopicConfigurer

protected Consumer<RetryTopicConfigurer> configureRetryTopicConfigurer()

Override this method if you need to configure the RetryTopicConfigurer.

Returns:

a RetryTopicConfigurer consumer.

configureDeadLetterPublishingContainerFactory

protected Consumer<DeadLetterPublishingRecovererFactory> configureDeadLetterPublishingContainerFactory()

Override this method to further configure the DeadLetterPublishingRecovererFactory.

Returns:

a DeadLetterPublishingRecovererFactory consumer.

configureListenerContainerFactoryConfigurer

protected Consumer<ListenerContainerFactoryConfigurer> configureListenerContainerFactoryConfigurer()

Override this method to further configure the ListenerContainerFactoryConfigurer.

Returns:

a ListenerContainerFactoryConfigurer consumer.

configureBlockingRetries

protected void configureBlockingRetries(RetryTopicConfigurationSupport.BlockingRetriesConfigurer blockingRetries)

Override this method to configure blocking retries parameters such as exceptions to be retried and the BackOff to be used.

Parameters:

blockingRetries - a RetryTopicConfigurationSupport.BlockingRetriesConfigurer.

manageNonBlockingFatalExceptions

protected void manageNonBlockingFatalExceptions(List<Class<? extends Throwable>> nonBlockingRetriesExceptions)

Override this method to manage non-blocking retries fatal exceptions. Records which processing throws an exception present in this list will be forwarded directly to the DLT, if one is configured, or stop being processed otherwise.

Parameters:

nonBlockingRetriesExceptions - a List of fatal exceptions containing the framework defaults.

configureCustomizers

protected void configureCustomizers(RetryTopicConfigurationSupport.CustomizersConfigurer customizersConfigurer)

Override this method to configure customizers for components created by non-blocking retries' configuration, such as MessageListenerContainer, DeadLetterPublishingRecoverer and DefaultErrorHandler.

Parameters:

customizersConfigurer - a RetryTopicConfigurationSupport.CustomizersConfigurer.

destinationTopicResolver

@Bean(name="org.springframework.kafka.retrytopic.internalDestinationTopicResolver")
public DestinationTopicResolver destinationTopicResolver(ObjectProvider<RetryTopicComponentFactory> componentFactoryProvider)

Return a global DestinationTopicResolver for resolving the DestinationTopic to which a given ConsumerRecord should be sent for retry. To configure it, consider overriding one of these other more fine-grained methods:

• manageNonBlockingFatalExceptions(java.util.List<java.lang.Class<? extends java.lang.Throwable>>) to configure non-blocking retries.
• configureDestinationTopicResolver() to further customize the component.
• createComponentFactory() to provide a subclass instance.

Parameters:

componentFactoryProvider - the component factory provider.

Returns:

the instance.

configureDestinationTopicResolver

protected Consumer<DestinationTopicResolver> configureDestinationTopicResolver()

Override this method to configure the DestinationTopicResolver.

Returns:

a DestinationTopicResolver consumer.

kafkaConsumerBackoffManager

@Bean(name="org.springframework.kafka.config.internalKafkaConsumerBackOffManager")
public KafkaConsumerBackoffManager kafkaConsumerBackoffManager(ApplicationContext applicationContext,
 @Qualifier("org.springframework.kafka.config.internalKafkaListenerEndpointRegistry") @Nullable
 ListenerContainerRegistry registry,
 ObjectProvider<RetryTopicComponentFactory> componentFactoryProvider,
 @Nullable
 RetryTopicSchedulerWrapper wrapper,
 @Nullable
 TaskScheduler taskScheduler)

Create the KafkaConsumerBackoffManager instance that will be used to back off partitions. To provide a custom implementation, either override this method, or override the RetryTopicComponentFactory.kafkaBackOffManagerFactory(org.springframework.kafka.listener.ListenerContainerRegistry, org.springframework.context.ApplicationContext) method and return a different KafkaBackOffManagerFactory.

Parameters:

applicationContext - the application context.

registry - the ListenerContainerRegistry to be used to fetch the MessageListenerContainer at runtime to be backed off.

componentFactoryProvider - the component factory provider.

wrapper - a RetryTopicSchedulerWrapper.

taskScheduler - a TaskScheduler.

Returns:

the instance.

createComponentFactory

protected RetryTopicComponentFactory createComponentFactory()

Override this method to provide a subclass of RetryTopicComponentFactory with different component implementations or subclasses.

Returns:

the instance.