Package org.springframework.kafka.listener

Class AbstractMessageListenerContainer<K,V>

java.lang.Object

org.springframework.kafka.listener.AbstractMessageListenerContainer<K,V>

Type Parameters:

K - the key type.

V - the value type.

All Implemented Interfaces:

Aware, BeanNameAware, DisposableBean, ApplicationContextAware, ApplicationEventPublisherAware, Lifecycle, Phased, SmartLifecycle, GenericMessageListenerContainer<K,V>, MessageListenerContainer

Direct Known Subclasses:

ConcurrentMessageListenerContainer, KafkaMessageListenerContainer

public abstract class AbstractMessageListenerContainer<K,V>
extends Object
implements GenericMessageListenerContainer<K,V>, BeanNameAware, ApplicationEventPublisherAware, ApplicationContextAware

The base implementation for the MessageListenerContainer.

Author:

Gary Russell, Marius Bogoevici, Artem Bilan, Tomaz Fernandes

Field Summary

Fields

Modifier and Type	Field	Description
protected final ConsumerFactory<K,V>	consumerFactory
static final int	DEFAULT_PHASE	The default SmartLifecycle phase for listener containers 2147483547.
protected final Object	lifecycleMonitor
protected final LogAccessor	logger

Constructor Summary

Constructors

Modifier	Constructor	Description
protected	AbstractMessageListenerContainer(ConsumerFactory<? super K,? super V> consumerFactory, ContainerProperties containerProperties)	Construct an instance with the provided factory and properties.

Method Summary

Modifier and Type	Method	Description
void	checkGroupId()
protected void	checkTopics()
protected final org.apache.kafka.clients.consumer.ConsumerRebalanceListener	createSimpleLoggingConsumerRebalanceListener()	Return default implementation of ConsumerRebalanceListener instance.
protected abstract void	doStart()
protected void	doStop(Runnable callback)
protected abstract void	doStop(Runnable callback, boolean normal)	Stop the container normally or abnormally.
AfterRollbackProcessor<? super K,? super V>	getAfterRollbackProcessor()	Return the currently configured AfterRollbackProcessor.
protected ApplicationContext	getApplicationContext()
ApplicationEventPublisher	getApplicationEventPublisher()	Get the event publisher.
protected BatchInterceptor<K,V>	getBatchInterceptor()
String	getBeanName()	Return the bean name.
CommonErrorHandler	getCommonErrorHandler()	Get the CommonErrorHandler.
ContainerProperties	getContainerProperties()	Return the container properties for this container.
GenericErrorHandler<?>	getGenericErrorHandler()	Deprecated. in favor of getCommonErrorHandler()
String	getGroupId()	Return the group.id property for this container whether specifically set on the container or via a consumer property on the consumer factory.
String	getListenerId()	The 'id' attribute of a @KafkaListener or the bean name for spring-managed containers.
byte[]	getListenerInfo()	Get arbitrary static information that will be added to the KafkaHeaders.LISTENER_INFO header of all records.
int	getPhase()
protected RecordInterceptor<K,V>	getRecordInterceptor()
boolean	isAutoStartup()
protected boolean	isInterceptBeforeTx()
boolean	isPartitionPauseRequested(org.apache.kafka.common.TopicPartition topicPartition)	Whether or not this topic's partition pause has been requested.
protected boolean	isPaused()
boolean	isPauseRequested()	Return true if MessageListenerContainer.pause() has been called; the container might not have actually paused yet.
boolean	isRunning()
protected boolean	isStoppedNormally()
protected AbstractMessageListenerContainer<?,?>	parentOrThis()	Return this or a parent container if this has a parent.
void	pause()	Pause this container before the next poll().
void	pausePartition(org.apache.kafka.common.TopicPartition topicPartition)	Pause this partition before the next poll().
protected void	publishContainerStoppedEvent()
void	resume()	Resume this container, if paused, after the next poll().
void	resumePartition(org.apache.kafka.common.TopicPartition topicPartition)	Resume this partition, if paused, after the next poll().
void	setAfterRollbackProcessor(AfterRollbackProcessor<? super K,? super V> afterRollbackProcessor)	Set a processor to perform seeks on unprocessed records after a rollback.
void	setApplicationContext(ApplicationContext applicationContext)
void	setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher)
void	setAutoStartup(boolean autoStartup)	Set the autoStartup.
void	setBatchErrorHandler(BatchErrorHandler errorHandler)	Deprecated. in favor of setCommonErrorHandler(CommonErrorHandler)
void	setBatchInterceptor(BatchInterceptor<K,V> batchInterceptor)	Set an interceptor to be called before calling the record listener.
void	setBeanName(String name)
void	setCommonErrorHandler(CommonErrorHandler commonErrorHandler)	Set the CommonErrorHandler which can handle errors for both record and batch listeners.
void	setErrorHandler(ErrorHandler errorHandler)	Deprecated. in favor of setCommonErrorHandler(CommonErrorHandler)
void	setGenericErrorHandler(GenericErrorHandler<?> errorHandler)	Deprecated. in favor of setCommonErrorHandler(CommonErrorHandler)
void	setInterceptBeforeTx(boolean interceptBeforeTx)	When false, invoke the interceptor after the transaction starts.
void	setListenerInfo(byte[] listenerInfo)	Set arbitrary information that will be added to the KafkaHeaders.LISTENER_INFO header of all records.
void	setPhase(int phase)
void	setRecordInterceptor(RecordInterceptor<K,V> recordInterceptor)	Set an interceptor to be called before calling the record listener.
protected void	setRunning(boolean running)
protected void	setStoppedNormally(boolean stoppedNormally)
void	setTopicCheckTimeout(int topicCheckTimeout)	How long to wait for Admin.describeTopics(Collection) result futures to complete.
void	setupMessageListener(Object messageListener)	Setup the message listener to use.
final void	start()
final void	stop()
final void	stop(boolean wait)	Stop the container.
void	stop(Runnable callback)
void	stopAbnormally(Runnable callback)	Stop the container after some exception so that MessageListenerContainer.isInExpectedState() will return false.

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.springframework.kafka.listener.MessageListenerContainer

destroy, getAssignedPartitions, getAssignmentsByClientId, isChildRunning, isContainerPaused, isInExpectedState, isPartitionPaused, metrics

Field Details

DEFAULT_PHASE

public static final int DEFAULT_PHASE

The default SmartLifecycle phase for listener containers 2147483547.

See Also:

• Constant Field Values

logger

protected final LogAccessor logger

consumerFactory

protected final ConsumerFactory<K,V> consumerFactory

lifecycleMonitor

protected final Object lifecycleMonitor

Constructor Details

AbstractMessageListenerContainer

protected AbstractMessageListenerContainer(ConsumerFactory<? super K,? super V> consumerFactory,
 ContainerProperties containerProperties)

Construct an instance with the provided factory and properties.

Parameters:

consumerFactory - the factory.

containerProperties - the properties.

Method Details

setApplicationContext

public void setApplicationContext(ApplicationContext applicationContext)
                           throws BeansException

Specified by:

setApplicationContext in interface ApplicationContextAware

Throws:

BeansException

getApplicationContext

@Nullable
protected ApplicationContext getApplicationContext()

setBeanName

public void setBeanName(String name)

Specified by:

setBeanName in interface BeanNameAware

getBeanName

@Nullable
public String getBeanName()

Return the bean name.

Returns:

the bean name.

setApplicationEventPublisher

public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher)

Specified by:

setApplicationEventPublisher in interface ApplicationEventPublisherAware

getApplicationEventPublisher

@Nullable
public ApplicationEventPublisher getApplicationEventPublisher()

Get the event publisher.

Returns:

the publisher

setErrorHandler

@Deprecated
public void setErrorHandler(ErrorHandler errorHandler)

Deprecated.

in favor of setCommonErrorHandler(CommonErrorHandler)

Set the error handler to call when the listener throws an exception.

Parameters:

errorHandler - the error handler.

Since:

2.2

See Also:

• setCommonErrorHandler(CommonErrorHandler)

setGenericErrorHandler

@Deprecated
public void setGenericErrorHandler(@Nullable
 GenericErrorHandler<?> errorHandler)

Deprecated.

in favor of setCommonErrorHandler(CommonErrorHandler)

Set the error handler to call when the listener throws an exception.

Parameters:

errorHandler - the error handler.

Since:

2.2

See Also:

• setCommonErrorHandler(CommonErrorHandler)

setBatchErrorHandler

@Deprecated
public void setBatchErrorHandler(BatchErrorHandler errorHandler)

Deprecated.

in favor of setCommonErrorHandler(CommonErrorHandler)

Set the batch error handler to call when the listener throws an exception.

Parameters:

errorHandler - the error handler.

Since:

2.2

See Also:

• setCommonErrorHandler(CommonErrorHandler)

getGenericErrorHandler

@Deprecated
@Nullable
public GenericErrorHandler<?> getGenericErrorHandler()

Deprecated.

in favor of getCommonErrorHandler()

Get the configured error handler.

Returns:

the error handler.

Since:

2.2

See Also:

• getCommonErrorHandler()

getCommonErrorHandler

@Nullable
public CommonErrorHandler getCommonErrorHandler()

Get the CommonErrorHandler.

Returns:

the handler.

Since:

2.8

setCommonErrorHandler

public void setCommonErrorHandler(@Nullable
 CommonErrorHandler commonErrorHandler)

Set the CommonErrorHandler which can handle errors for both record and batch listeners. Replaces the use of GenericErrorHandlers.

Parameters:

commonErrorHandler - the handler.

Since:

2.8

isStoppedNormally

protected boolean isStoppedNormally()

setStoppedNormally

protected void setStoppedNormally(boolean stoppedNormally)

isAutoStartup

public boolean isAutoStartup()

Specified by:

isAutoStartup in interface SmartLifecycle

setAutoStartup

public void setAutoStartup(boolean autoStartup)

Description copied from interface: MessageListenerContainer

Set the autoStartup.

Specified by:

setAutoStartup in interface MessageListenerContainer

Parameters:

autoStartup - the autoStartup to set.

See Also:

• SmartLifecycle

setRunning

protected void setRunning(boolean running)

isRunning

public boolean isRunning()

Specified by:

isRunning in interface Lifecycle

isPaused

protected boolean isPaused()

isPartitionPauseRequested

public boolean isPartitionPauseRequested(org.apache.kafka.common.TopicPartition topicPartition)

Description copied from interface: MessageListenerContainer

Whether or not this topic's partition pause has been requested.

Specified by:

isPartitionPauseRequested in interface MessageListenerContainer

Parameters:

topicPartition - the topic partition to check

Returns:

true if pause for this TopicPartition has been requested

pausePartition

public void pausePartition(org.apache.kafka.common.TopicPartition topicPartition)

Description copied from interface: MessageListenerContainer

Pause this partition before the next poll(). This is a thread-safe operation, the actual pause is processed by the consumer thread.

Specified by:

pausePartition in interface MessageListenerContainer

Parameters:

topicPartition - the topicPartition to pause.

resumePartition

public void resumePartition(org.apache.kafka.common.TopicPartition topicPartition)

Description copied from interface: MessageListenerContainer

Resume this partition, if paused, after the next poll(). This is a thread-safe operation, the actual pause is processed by the consumer thread.

Specified by:

resumePartition in interface MessageListenerContainer

Parameters:

topicPartition - the topicPartition to resume.

isPauseRequested

public boolean isPauseRequested()

Description copied from interface: MessageListenerContainer

Return true if MessageListenerContainer.pause() has been called; the container might not have actually paused yet.

Specified by:

isPauseRequested in interface MessageListenerContainer

Returns:

true if pause has been requested.

setPhase

public void setPhase(int phase)

getPhase

public int getPhase()

Specified by:

getPhase in interface Phased

Specified by:

getPhase in interface SmartLifecycle

getAfterRollbackProcessor

public AfterRollbackProcessor<? super K,? super V> getAfterRollbackProcessor()

Return the currently configured AfterRollbackProcessor.

Returns:

the after rollback processor.

Since:

2.2.14

setAfterRollbackProcessor

public void setAfterRollbackProcessor(AfterRollbackProcessor<? super K,? super V> afterRollbackProcessor)

Set a processor to perform seeks on unprocessed records after a rollback. Default will seek to current position all topics/partitions, including the failed record.

Parameters:

afterRollbackProcessor - the processor.

Since:

1.3.5

getContainerProperties

public ContainerProperties getContainerProperties()

Description copied from interface: MessageListenerContainer

Return the container properties for this container.

Specified by:

getContainerProperties in interface MessageListenerContainer

Returns:

the properties.

getGroupId

@Nullable
public String getGroupId()

Description copied from interface: MessageListenerContainer

Return the group.id property for this container whether specifically set on the container or via a consumer property on the consumer factory.

Specified by:

getGroupId in interface MessageListenerContainer

Returns:

the group id.

getListenerId

@Nullable
public String getListenerId()

Description copied from interface: MessageListenerContainer

The 'id' attribute of a @KafkaListener or the bean name for spring-managed containers.

Specified by:

getListenerId in interface MessageListenerContainer

Returns:

the id or bean name.

getListenerInfo

@Nullable
public byte[] getListenerInfo()

Get arbitrary static information that will be added to the KafkaHeaders.LISTENER_INFO header of all records.

Returns:

the info.

Since:

2.8.4

setListenerInfo

public void setListenerInfo(@Nullable
 byte[] listenerInfo)

Set arbitrary information that will be added to the KafkaHeaders.LISTENER_INFO header of all records.

Parameters:

listenerInfo - the info.

Since:

2.8.4

setTopicCheckTimeout

public void setTopicCheckTimeout(int topicCheckTimeout)

How long to wait for Admin.describeTopics(Collection) result futures to complete.

Parameters:

topicCheckTimeout - the timeout in seconds; default 30.

Since:

2.3

getRecordInterceptor

protected RecordInterceptor<K,V> getRecordInterceptor()

setRecordInterceptor

public void setRecordInterceptor(RecordInterceptor<K,V> recordInterceptor)

Set an interceptor to be called before calling the record listener. Does not apply to batch listeners.

Parameters:

recordInterceptor - the interceptor.

Since:

2.2.7

See Also:

• setInterceptBeforeTx(boolean)

getBatchInterceptor

protected BatchInterceptor<K,V> getBatchInterceptor()

setBatchInterceptor

public void setBatchInterceptor(BatchInterceptor<K,V> batchInterceptor)

Set an interceptor to be called before calling the record listener.

Parameters:

batchInterceptor - the interceptor.

Since:

2.6.6

See Also:

• setInterceptBeforeTx(boolean)

isInterceptBeforeTx

protected boolean isInterceptBeforeTx()

setInterceptBeforeTx

public void setInterceptBeforeTx(boolean interceptBeforeTx)

When false, invoke the interceptor after the transaction starts.

Parameters:

interceptBeforeTx - false to intercept within the transaction. Default true since 2.8.

Since:

2.3.4

See Also:

• setRecordInterceptor(RecordInterceptor)
• setBatchInterceptor(BatchInterceptor)

setupMessageListener

public void setupMessageListener(Object messageListener)

Description copied from interface: MessageListenerContainer

Setup the message listener to use. Throws an IllegalArgumentException if that message listener type is not supported.

Specified by:

setupMessageListener in interface MessageListenerContainer

Parameters:

messageListener - the object to wrapped to the MessageListener.

start

public final void start()

Specified by:

start in interface Lifecycle

checkTopics

protected void checkTopics()

checkGroupId

public void checkGroupId()

doStart

protected abstract void doStart()

stop

public final void stop()

Specified by:

stop in interface Lifecycle

stop

public final void stop(boolean wait)

Stop the container.

Parameters:

wait - wait for the listener to terminate.

Since:

2.3.8

pause

public void pause()

Description copied from interface: MessageListenerContainer

Pause this container before the next poll(). This is a thread-safe operation, the actual pause is processed by the consumer thread.

Specified by:

pause in interface MessageListenerContainer

See Also:

• KafkaConsumer.pause(Collection)

resume

public void resume()

Description copied from interface: MessageListenerContainer

Resume this container, if paused, after the next poll(). This is a thread-safe operation, the actual resume is processed by the consumer thread.

Specified by:

resume in interface MessageListenerContainer

See Also:

• KafkaConsumer.resume(Collection)

stop

public void stop(Runnable callback)

Specified by:

stop in interface SmartLifecycle

stopAbnormally

public void stopAbnormally(Runnable callback)

Description copied from interface: MessageListenerContainer

Stop the container after some exception so that MessageListenerContainer.isInExpectedState() will return false.

Specified by:

stopAbnormally in interface MessageListenerContainer

Parameters:

callback - the callback.

See Also:

• MessageListenerContainer.isInExpectedState()

doStop

protected void doStop(Runnable callback)

doStop

protected abstract void doStop(Runnable callback,
 boolean normal)

Stop the container normally or abnormally.

Parameters:

callback - the callback.

normal - true for an expected stop.

Since:

2.8

createSimpleLoggingConsumerRebalanceListener

protected final org.apache.kafka.clients.consumer.ConsumerRebalanceListener createSimpleLoggingConsumerRebalanceListener()

Return default implementation of ConsumerRebalanceListener instance.

Returns:

the ConsumerRebalanceListener currently assigned to this container.

publishContainerStoppedEvent

protected void publishContainerStoppedEvent()

parentOrThis

protected AbstractMessageListenerContainer<?,?> parentOrThis()

Return this or a parent container if this has a parent.

Returns:

the parent or this.

Since:

2.2.1