org.springframework.kafka.config

Class AbstractKafkaListenerContainerFactory<C extends AbstractMessageListenerContainer<K,V>,K,V>

java.lang.Object

org.springframework.kafka.config.AbstractKafkaListenerContainerFactory<C,K,V>

Type Parameters:

C - the AbstractMessageListenerContainer implementation type.

K - the key type.

V - the value type.

All Implemented Interfaces:

org.springframework.beans.factory.Aware, org.springframework.beans.factory.InitializingBean, org.springframework.context.ApplicationContextAware, org.springframework.context.ApplicationEventPublisherAware, KafkaListenerContainerFactory<C>

Direct Known Subclasses:

ConcurrentKafkaListenerContainerFactory

public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMessageListenerContainer<K,V>,K,V>
extends java.lang.Object
implements KafkaListenerContainerFactory<C>, org.springframework.context.ApplicationEventPublisherAware, org.springframework.beans.factory.InitializingBean, org.springframework.context.ApplicationContextAware

Base KafkaListenerContainerFactory for Spring's base container implementation.

Author:

Stephane Nicoll, Gary Russell, Artem Bilan

See Also:

AbstractMessageListenerContainer

Field Summary

Fields

Modifier and Type	Field and Description
protected org.springframework.core.log.LogAccessor	logger

Constructor Summary

Constructors

Constructor and Description
AbstractKafkaListenerContainerFactory()

Method Summary

Modifier and Type	Method and Description
void	afterPropertiesSet()
C	createContainer(java.util.regex.Pattern topicPattern) Create and configure a container without a listener; used to create containers that are not used for KafkaListener annotations.
C	createContainer(java.lang.String... topics) Create and configure a container without a listener; used to create containers that are not used for KafkaListener annotations.
C	createContainer(TopicPartitionOffset... topicsAndPartitions) Create and configure a container without a listener; used to create containers that are not used for KafkaListener annotations.
protected abstract C	createContainerInstance(KafkaListenerEndpoint endpoint) Create an empty container instance.
C	createListenerContainer(KafkaListenerEndpoint endpoint) Create a MessageListenerContainer for the given KafkaListenerEndpoint.
ConsumerFactory<? super K,? super V>	getConsumerFactory()
ContainerProperties	getContainerProperties() Obtain the properties template for this factory - set properties as needed and they will be copied to a final properties instance for the endpoint.
protected void	initializeContainer(C instance, KafkaListenerEndpoint endpoint) Further initialize the specified container.
java.lang.Boolean	isBatchListener() Return true if this endpoint creates a batch listener.
void	setAckDiscarded(java.lang.Boolean ackDiscarded) Set to true to ack discards when a filter strategy is in use.
void	setAfterRollbackProcessor(AfterRollbackProcessor<? super K,? super V> afterRollbackProcessor) Set a processor to invoke after a transaction rollback; typically will seek the unprocessed topic/partition to reprocess the records.
void	setApplicationContext(org.springframework.context.ApplicationContext applicationContext)
void	setApplicationEventPublisher(org.springframework.context.ApplicationEventPublisher applicationEventPublisher)
void	setAutoStartup(java.lang.Boolean autoStartup) Specify an autoStartup boolean flag.
void	setBatchErrorHandler(BatchErrorHandler errorHandler) Set the batch error handler to call when the listener throws an exception.
void	setBatchListener(java.lang.Boolean batchListener) Set to true if this endpoint should create a batch listener.
void	setBatchToRecordAdapter(BatchToRecordAdapter<K,V> batchToRecordAdapter) Set a BatchToRecordAdapter.
void	setConsumerFactory(ConsumerFactory<? super K,? super V> consumerFactory) Specify a ConsumerFactory to use.
void	setContainerCustomizer(ContainerCustomizer<K,V,C> containerCustomizer) Set a customizer used to further configure a container after it has been created.
void	setErrorHandler(ErrorHandler errorHandler) Set the error handler to call when the listener throws an exception.
void	setMessageConverter(MessageConverter messageConverter) Set the message converter to use if dynamic argument type matching is needed.
void	setMissingTopicsFatal(boolean missingTopicsFatal) Set to false to allow the container to start even if any of the configured topics are not present on the broker.
void	setPhase(int phase) Specify a phase to use.
void	setRecordFilterStrategy(RecordFilterStrategy<? super K,? super V> recordFilterStrategy) Set the record filter strategy.
void	setRecordInterceptor(RecordInterceptor<K,V> recordInterceptor) Set an interceptor to be called before calling the listener.
void	setRecoveryCallback(org.springframework.retry.RecoveryCallback<? extends java.lang.Object> recoveryCallback) Set a callback to be used with the retryTemplate.
void	setReplyHeadersConfigurer(ReplyHeadersConfigurer replyHeadersConfigurer) Set a configurer which will be invoked when creating a reply message.
void	setReplyTemplate(KafkaTemplate<?,?> replyTemplate) Set the KafkaTemplate to use to send replies.
void	setRetryTemplate(org.springframework.retry.support.RetryTemplate retryTemplate) Set a retryTemplate.
void	setStatefulRetry(boolean statefulRetry) When using a RetryTemplate Set to true to enable stateful retry.

Methods inherited from class java.lang.Object

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

Field Detail

logger

protected final org.springframework.core.log.LogAccessor logger

Constructor Detail

AbstractKafkaListenerContainerFactory

public AbstractKafkaListenerContainerFactory()

Method Detail

setApplicationContext

public void setApplicationContext(org.springframework.context.ApplicationContext applicationContext)
                           throws org.springframework.beans.BeansException

Specified by:

setApplicationContext in interface org.springframework.context.ApplicationContextAware

Throws:

org.springframework.beans.BeansException

setConsumerFactory

public void setConsumerFactory(ConsumerFactory<? super K,? super V> consumerFactory)

Specify a ConsumerFactory to use.

Parameters:

consumerFactory - The consumer factory.

getConsumerFactory

public ConsumerFactory<? super K,? super V> getConsumerFactory()

setAutoStartup

public void setAutoStartup(java.lang.Boolean autoStartup)

Specify an autoStartup boolean flag.

Parameters:

autoStartup - true for auto startup.

See Also:

AbstractMessageListenerContainer.setAutoStartup(boolean)

setPhase

public void setPhase(int phase)

Specify a phase to use.

Parameters:

phase - The phase.

See Also:

AbstractMessageListenerContainer.setPhase(int)

setMessageConverter

public void setMessageConverter(MessageConverter messageConverter)

Set the message converter to use if dynamic argument type matching is needed.

Parameters:

messageConverter - the converter.

setRecordFilterStrategy

public void setRecordFilterStrategy(RecordFilterStrategy<? super K,? super V> recordFilterStrategy)

Set the record filter strategy.

Parameters:

recordFilterStrategy - the strategy.

setAckDiscarded

public void setAckDiscarded(java.lang.Boolean ackDiscarded)

Set to true to ack discards when a filter strategy is in use.

Parameters:

ackDiscarded - the ackDiscarded.

setRetryTemplate

public void setRetryTemplate(org.springframework.retry.support.RetryTemplate retryTemplate)

Set a retryTemplate.

Parameters:

retryTemplate - the template.

setRecoveryCallback

public void setRecoveryCallback(org.springframework.retry.RecoveryCallback<? extends java.lang.Object> recoveryCallback)

Set a callback to be used with the retryTemplate.

Parameters:

recoveryCallback - the callback.

setStatefulRetry

public void setStatefulRetry(boolean statefulRetry)

When using a RetryTemplate Set to true to enable stateful retry. Use in conjunction with a SeekToCurrentErrorHandler when retry can take excessive time; each failure goes back to the broker, to keep the Consumer alive.

Parameters:

statefulRetry - true to enable stateful retry.

Since:

2.1.3

isBatchListener

public java.lang.Boolean isBatchListener()

Return true if this endpoint creates a batch listener.

Returns:

true for a batch listener.

Since:

1.1

setBatchListener

public void setBatchListener(java.lang.Boolean batchListener)

Set to true if this endpoint should create a batch listener.

Parameters:

batchListener - true for a batch listener.

Since:

1.1

setApplicationEventPublisher

public void setApplicationEventPublisher(org.springframework.context.ApplicationEventPublisher applicationEventPublisher)

Specified by:

setApplicationEventPublisher in interface org.springframework.context.ApplicationEventPublisherAware

setReplyTemplate

public void setReplyTemplate(KafkaTemplate<?,?> replyTemplate)

Set the KafkaTemplate to use to send replies.

Parameters:

replyTemplate - the template.

Since:

2.0

setErrorHandler

public void setErrorHandler(ErrorHandler errorHandler)

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

Parameters:

errorHandler - the error handler.

Since:

2.2

setBatchErrorHandler

public void setBatchErrorHandler(BatchErrorHandler errorHandler)

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

Parameters:

errorHandler - the error handler.

Since:

2.2

setAfterRollbackProcessor

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

Set a processor to invoke after a transaction rollback; typically will seek the unprocessed topic/partition to reprocess the records. The default does so, including the failed record.

Parameters:

afterRollbackProcessor - the processor.

Since:

1.3.5

setReplyHeadersConfigurer

public void setReplyHeadersConfigurer(ReplyHeadersConfigurer replyHeadersConfigurer)

Set a configurer which will be invoked when creating a reply message.

Parameters:

replyHeadersConfigurer - the configurer.

Since:

2.2

setMissingTopicsFatal

public void setMissingTopicsFatal(boolean missingTopicsFatal)

Set to false to allow the container to start even if any of the configured topics are not present on the broker. Does not apply when topic patterns are configured. Default true;

Parameters:

missingTopicsFatal - the missingTopicsFatal.

Since:

2.3

getContainerProperties

public ContainerProperties getContainerProperties()

Obtain the properties template for this factory - set properties as needed and they will be copied to a final properties instance for the endpoint.

Returns:

the properties.

setRecordInterceptor

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

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

Parameters:

recordInterceptor - the interceptor.

Since:

2.2.7

setBatchToRecordAdapter

public void setBatchToRecordAdapter(BatchToRecordAdapter<K,V> batchToRecordAdapter)

Set a BatchToRecordAdapter.

Parameters:

batchToRecordAdapter - the adapter.

Since:

2.4.2

setContainerCustomizer

public void setContainerCustomizer(ContainerCustomizer<K,V,C> containerCustomizer)

Set a customizer used to further configure a container after it has been created.

Parameters:

containerCustomizer - the customizer.

Since:

2.3.4

afterPropertiesSet

public void afterPropertiesSet()

Specified by:

afterPropertiesSet in interface org.springframework.beans.factory.InitializingBean

createListenerContainer

public C createListenerContainer(KafkaListenerEndpoint endpoint)

Description copied from interface: KafkaListenerContainerFactory

Create a MessageListenerContainer for the given KafkaListenerEndpoint. Containers created using this method are added to the listener endpoint registry.

Specified by:

createListenerContainer in interface KafkaListenerContainerFactory<C extends AbstractMessageListenerContainer<K,V>>

Parameters:

endpoint - the endpoint to configure

Returns:

the created container

createContainerInstance

protected abstract C createContainerInstance(KafkaListenerEndpoint endpoint)

Create an empty container instance.

Parameters:

endpoint - the endpoint.

Returns:

the new container instance.

initializeContainer

protected void initializeContainer(C instance,
                                   KafkaListenerEndpoint endpoint)

Further initialize the specified container.

Subclasses can inherit from this method to apply extra configuration if necessary.

Parameters:

instance - the container instance to configure.

endpoint - the endpoint.

createContainer

public C createContainer(TopicPartitionOffset... topicsAndPartitions)

Description copied from interface: KafkaListenerContainerFactory

Create and configure a container without a listener; used to create containers that are not used for KafkaListener annotations. Containers created using this method are not added to the listener endpoint registry.

Specified by:

createContainer in interface KafkaListenerContainerFactory<C extends AbstractMessageListenerContainer<K,V>>

Parameters:

topicsAndPartitions - the topicPartitions to assign.

Returns:

the container.

createContainer

public C createContainer(java.lang.String... topics)

Description copied from interface: KafkaListenerContainerFactory

Create and configure a container without a listener; used to create containers that are not used for KafkaListener annotations. Containers created using this method are not added to the listener endpoint registry.

Specified by:

createContainer in interface KafkaListenerContainerFactory<C extends AbstractMessageListenerContainer<K,V>>

Parameters:

topics - the topics.

Returns:

the container.

createContainer

public C createContainer(java.util.regex.Pattern topicPattern)

Description copied from interface: KafkaListenerContainerFactory

Create and configure a container without a listener; used to create containers that are not used for KafkaListener annotations. Containers created using this method are not added to the listener endpoint registry.

Specified by:

createContainer in interface KafkaListenerContainerFactory<C extends AbstractMessageListenerContainer<K,V>>

Parameters:

topicPattern - the topicPattern.

Returns:

the container.