Package org.springframework.kafka.annotation

Class KafkaListenerAnnotationBeanPostProcessor<K,V>

java.lang.Object

org.springframework.kafka.annotation.KafkaListenerAnnotationBeanPostProcessor<K,V>

Type Parameters:

K - the key type.

V - the value type.

All Implemented Interfaces:

Aware, BeanPostProcessor, InitializingBean, SmartInitializingSingleton, ApplicationContextAware, Ordered

public class KafkaListenerAnnotationBeanPostProcessor<K,V>
extends Object
implements BeanPostProcessor, Ordered, ApplicationContextAware, InitializingBean, SmartInitializingSingleton

Bean post-processor that registers methods annotated with KafkaListener to be invoked by a Kafka message listener container created under the covers by a KafkaListenerContainerFactory according to the parameters of the annotation.

Annotated methods can use flexible arguments as defined by KafkaListener.

This post-processor is automatically registered by Spring's EnableKafka annotation.

Auto-detect any KafkaListenerConfigurer instances in the container, allowing for customization of the registry to be used, the default container factory or for fine-grained control over endpoints registration. See EnableKafka Javadoc for complete usage details.

Author:

Stephane Nicoll, Juergen Hoeller, Gary Russell, Artem Bilan, Dariusz Szablinski, Venil Noronha, Dimitri Penner, Filip Halemba, Tomaz Fernandes, Wang Zhiyang

See Also:

• KafkaListener
• KafkaListenerErrorHandler
• EnableKafka
• KafkaListenerConfigurer
• KafkaListenerEndpointRegistrar
• KafkaListenerEndpointRegistry
• KafkaListenerEndpoint
• MethodKafkaListenerEndpoint

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static interface	KafkaListenerAnnotationBeanPostProcessor.AnnotationEnhancer	Post processes each set of annotation attributes.

Field Summary

Fields

Modifier and Type	Field	Description
static final String	DEFAULT_KAFKA_LISTENER_CONTAINER_FACTORY_BEAN_NAME	The bean name of the default KafkaListenerContainerFactory.

Fields inherited from interface org.springframework.core.Ordered

HIGHEST_PRECEDENCE, LOWEST_PRECEDENCE

Constructor Summary

Constructors

Constructor	Description
KafkaListenerAnnotationBeanPostProcessor()

Method Summary

Modifier and Type	Method	Description
void	afterPropertiesSet()
void	afterSingletonsInstantiated()
protected void	assertBeanFactory()
KafkaListenerEndpointRegistrar	getEndpointRegistrar()	Return the KafkaListenerEndpointRegistrar.
MessageHandlerMethodFactory	getMessageHandlerMethodFactory()	Return the configured handler factory.
int	getOrder()
protected String	noBeanFoundMessage(Object target, String listenerBeanName, String requestedBeanName, Class<?> expectedClass)
Object	postProcessAfterInitialization(Object bean, String beanName)
Object	postProcessBeforeInitialization(Object bean, String beanName)
protected void	processKafkaListener(KafkaListener kafkaListener, Method method, Object bean, String beanName)
protected void	processListener(MethodKafkaListenerEndpoint<?,?> endpoint, KafkaListener kafkaListener, Object bean, String beanName, String[] topics, TopicPartitionOffset[] tps)
void	setApplicationContext(ApplicationContext applicationContext)
void	setBeanFactory(BeanFactory beanFactory)	Making a BeanFactory available is optional; if not set, KafkaListenerConfigurer beans won't get auto-detected and an endpoint registry has to be explicitly configured.
void	setCharset(Charset charset)	Set a charset to use when converting byte[] to String in method arguments and other String/byte[] conversions.
void	setDefaultContainerFactoryBeanName(String containerFactoryBeanName)	Set the name of the KafkaListenerContainerFactory to use by default.
void	setEndpointRegistry(KafkaListenerEndpointRegistry endpointRegistry)	Set the KafkaListenerEndpointRegistry that will hold the created endpoint and manage the lifecycle of the related listener container.
void	setMessageHandlerMethodFactory(MessageHandlerMethodFactory messageHandlerMethodFactory)	Set the MessageHandlerMethodFactory to use to configure the message listener responsible to serve an endpoint detected by this processor.

Methods inherited from class java.lang.Object

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

Field Details

DEFAULT_KAFKA_LISTENER_CONTAINER_FACTORY_BEAN_NAME

public static final String DEFAULT_KAFKA_LISTENER_CONTAINER_FACTORY_BEAN_NAME

The bean name of the default KafkaListenerContainerFactory.

See Also:

• Constant Field Values

Constructor Details

KafkaListenerAnnotationBeanPostProcessor

public KafkaListenerAnnotationBeanPostProcessor()

Method Details

getOrder

public int getOrder()

Specified by:

getOrder in interface Ordered

setEndpointRegistry

public void setEndpointRegistry(KafkaListenerEndpointRegistry endpointRegistry)

Set the KafkaListenerEndpointRegistry that will hold the created endpoint and manage the lifecycle of the related listener container.

Parameters:

endpointRegistry - the KafkaListenerEndpointRegistry to set.

setDefaultContainerFactoryBeanName

public void setDefaultContainerFactoryBeanName(String containerFactoryBeanName)

Set the name of the KafkaListenerContainerFactory to use by default.

If none is specified, "kafkaListenerContainerFactory" is assumed to be defined.

Parameters:

containerFactoryBeanName - the KafkaListenerContainerFactory bean name.

setMessageHandlerMethodFactory

public void setMessageHandlerMethodFactory(MessageHandlerMethodFactory messageHandlerMethodFactory)

Set the MessageHandlerMethodFactory to use to configure the message listener responsible to serve an endpoint detected by this processor.

By default, DefaultMessageHandlerMethodFactory is used and it can be configured further to support additional method arguments or to customize conversion and validation support. See DefaultMessageHandlerMethodFactory Javadoc for more details.

Parameters:

messageHandlerMethodFactory - the MessageHandlerMethodFactory instance.

getMessageHandlerMethodFactory

public MessageHandlerMethodFactory getMessageHandlerMethodFactory()

Return the configured handler factory.

Returns:

the factory.

Since:

2.5.7

getEndpointRegistrar

public KafkaListenerEndpointRegistrar getEndpointRegistrar()

Return the KafkaListenerEndpointRegistrar.

Returns:

the registrar.

Since:

2.9.3

setApplicationContext

public void setApplicationContext(ApplicationContext applicationContext)
                           throws BeansException

Specified by:

setApplicationContext in interface ApplicationContextAware

Throws:

BeansException

setBeanFactory

public void setBeanFactory(BeanFactory beanFactory)

Making a BeanFactory available is optional; if not set, KafkaListenerConfigurer beans won't get auto-detected and an endpoint registry has to be explicitly configured.

Parameters:

beanFactory - the BeanFactory to be used.

setCharset

public void setCharset(Charset charset)

Set a charset to use when converting byte[] to String in method arguments and other String/byte[] conversions. Default UTF-8.

Parameters:

charset - the charset.

Since:

2.2

afterPropertiesSet

public void afterPropertiesSet()
                        throws Exception

Specified by:

afterPropertiesSet in interface InitializingBean

Throws:

Exception

afterSingletonsInstantiated

public void afterSingletonsInstantiated()

Specified by:

afterSingletonsInstantiated in interface SmartInitializingSingleton

postProcessBeforeInitialization

public Object postProcessBeforeInitialization(Object bean,
 String beanName)
                                       throws BeansException

Specified by:

postProcessBeforeInitialization in interface BeanPostProcessor

Throws:

BeansException

postProcessAfterInitialization

public Object postProcessAfterInitialization(Object bean,
 String beanName)
                                      throws BeansException

Specified by:

postProcessAfterInitialization in interface BeanPostProcessor

Throws:

BeansException

processKafkaListener

protected void processKafkaListener(KafkaListener kafkaListener,
 Method method,
 Object bean,
 String beanName)

processListener

protected void processListener(MethodKafkaListenerEndpoint<?,?> endpoint,
 KafkaListener kafkaListener,
 Object bean,
 String beanName,
 String[] topics,
 TopicPartitionOffset[] tps)

assertBeanFactory

protected void assertBeanFactory()

noBeanFoundMessage

protected String noBeanFoundMessage(Object target,
 String listenerBeanName,
 String requestedBeanName,
 Class<?> expectedClass)