Package org.springframework.kafka.config

Class AbstractKafkaListenerEndpoint<K,V>

java.lang.Object

org.springframework.kafka.config.AbstractKafkaListenerEndpoint<K,V>

Type Parameters:

K - the key type.

V - the value type.

All Implemented Interfaces:

Aware, BeanFactoryAware, InitializingBean, KafkaListenerEndpoint

Direct Known Subclasses:

MethodKafkaListenerEndpoint

public abstract class AbstractKafkaListenerEndpoint<K,V>
extends Object
implements KafkaListenerEndpoint, BeanFactoryAware, InitializingBean

Base model for a Kafka listener endpoint.

Author:

Stephane Nicoll, Gary Russell, Artem Bilan

See Also:

• MethodKafkaListenerEndpoint

Constructor Summary

Constructors

Constructor	Description
AbstractKafkaListenerEndpoint()

Method Summary

Modifier and Type	Method	Description
void	afterPropertiesSet()
protected abstract MessagingMessageListenerAdapter<K,V>	createMessageListener(MessageListenerContainer container, MessageConverter messageConverter)	Create a MessageListener that is able to serve this endpoint for the specified container.
Boolean	getAutoStartup()	Return the autoStartup for this endpoint's container.
Boolean	getBatchListener()	Return the current batch listener flag for this endpoint, or null if not explicitly set.
protected BatchToRecordAdapter<K,V>	getBatchToRecordAdapter()
protected BeanExpressionContext	getBeanExpressionContext()
protected BeanFactory	getBeanFactory()
protected BeanResolver	getBeanResolver()
String	getClientIdPrefix()	Return the client id prefix for the container; it will be suffixed by '-n' to provide a unique id when concurrency is used.
Integer	getConcurrency()	Return the concurrency for this endpoint's container.
Properties	getConsumerProperties()	Get the consumer properties that will be merged with the consumer properties provided by the consumer factory; properties here will supersede any with the same name(s) in the consumer factory.
protected StringBuilder	getEndpointDescription()	Return a description for this endpoint.
String	getGroup()	Return the group of this endpoint or null if not in a group.
String	getGroupId()	Return the groupId of this endpoint - if present, overrides the group.id property of the consumer factory.
String	getId()	Return the id of this endpoint.
byte[]	getListenerInfo()	Get the listener info to insert in the record header.
String	getMainListenerId()	Return the main listener id if this container is for a retry topic.
protected RecordFilterStrategy<? super K,? super V>	getRecordFilterStrategy()
protected KafkaTemplate<?,?>	getReplyTemplate()
protected BeanExpressionResolver	getResolver()
TopicPartitionOffset[]	getTopicPartitionsToAssign()	Return the topicPartitions for this endpoint.
Pattern	getTopicPattern()	Return the topicPattern for this endpoint.
Collection<String>	getTopics()	Return the topics for this endpoint.
protected boolean	isAckDiscarded()
boolean	isBatchListener()	Return true if this endpoint creates a batch listener.
boolean	isSplitIterables()	When true, Iterable return results will be split into discrete records.
void	setAckDiscarded(boolean ackDiscarded)	Set to true if the setRecordFilterStrategy(RecordFilterStrategy) is in use.
void	setAutoStartup(Boolean autoStartup)	Set the autoStartup for this endpoint's container.
void	setBatchListener(boolean batchListener)	Set to true if this endpoint should create a batch listener.
void	setBatchToRecordAdapter(BatchToRecordAdapter<K,V> batchToRecordAdapter)	Set a BatchToRecordAdapter.
void	setBeanFactory(BeanFactory beanFactory)
void	setClientIdPrefix(String clientIdPrefix)	Set the client id prefix; overrides the client id in the consumer configuration properties.
void	setConcurrency(Integer concurrency)	Set the concurrency for this endpoint's container.
void	setConsumerProperties(Properties consumerProperties)	Set the consumer properties that will be merged with the consumer properties provided by the consumer factory; properties here will supersede any with the same name(s) in the consumer factory.
void	setCorrelationHeaderName(String correlationHeaderName)	Set a custom header name for the correlation id.
void	setGroup(String group)	Set the group for the corresponding listener container.
void	setGroupId(String groupId)	Set the group id to override the group.id property in the ContainerFactory.
void	setId(String id)
void	setListenerInfo(byte[] listenerInfo)	Set the listener info to insert in the record header.
void	setMainListenerId(String id)
void	setRecordFilterStrategy(RecordFilterStrategy<? super K,? super V> recordFilterStrategy)	Set a RecordFilterStrategy implementation.
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	setSplitIterables(boolean splitIterables)	Set to false to disable splitting Iterable reply values into separate records.
void	setTopicPartitions(TopicPartitionOffset... topicPartitions)	Set the topicPartitions to use.
void	setTopicPattern(Pattern topicPattern)	Set the topic pattern to use.
void	setTopics(String... topics)	Set the topics to use.
void	setupListenerContainer(MessageListenerContainer listenerContainer, MessageConverter messageConverter)	Setup the specified message listener container with the model defined by this endpoint.
String	toString()

Methods inherited from class java.lang.Object

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

Constructor Details

AbstractKafkaListenerEndpoint

public AbstractKafkaListenerEndpoint()

Method Details

setBeanFactory

public void setBeanFactory(BeanFactory beanFactory)
                    throws BeansException

Specified by:

setBeanFactory in interface BeanFactoryAware

Throws:

BeansException

getBeanFactory

@Nullable
protected BeanFactory getBeanFactory()

getResolver

@Nullable
protected BeanExpressionResolver getResolver()

getBeanExpressionContext

@Nullable
protected BeanExpressionContext getBeanExpressionContext()

getBeanResolver

@Nullable
protected BeanResolver getBeanResolver()

setId

public void setId(@Nullable
 String id)

setMainListenerId

public void setMainListenerId(@Nullable
 String id)

getMainListenerId

@Nullable
public String getMainListenerId()

Description copied from interface: KafkaListenerEndpoint

Return the main listener id if this container is for a retry topic.

Specified by:

getMainListenerId in interface KafkaListenerEndpoint

Returns:

the main listener id or null.

getId

@Nullable
public String getId()

Description copied from interface: KafkaListenerEndpoint

Return the id of this endpoint.

Specified by:

getId in interface KafkaListenerEndpoint

Returns:

the id of this endpoint. The id can be further qualified when the endpoint is resolved against its actual listener container.

See Also:

• KafkaListenerContainerFactory.createListenerContainer(org.springframework.kafka.config.KafkaListenerEndpoint)

setGroupId

public void setGroupId(String groupId)

Set the group id to override the group.id property in the ContainerFactory.

Parameters:

groupId - the group id.

Since:

1.3

getGroupId

@Nullable
public String getGroupId()

Description copied from interface: KafkaListenerEndpoint

Return the groupId of this endpoint - if present, overrides the group.id property of the consumer factory.

Specified by:

getGroupId in interface KafkaListenerEndpoint

Returns:

the group id; may be null.

setTopics

public void setTopics(String... topics)

Set the topics to use. Either these or 'topicPattern' or 'topicPartitions' should be provided, but not a mixture.

Parameters:

topics - to set.

See Also:

• setTopicPartitions(TopicPartitionOffset...)
• setTopicPattern(Pattern)

getTopics

public Collection<String> getTopics()

Return the topics for this endpoint.

Specified by:

getTopics in interface KafkaListenerEndpoint

Returns:

the topics for this endpoint.

setTopicPartitions

public void setTopicPartitions(TopicPartitionOffset... topicPartitions)

Set the topicPartitions to use. Either this or 'topic' or 'topicPattern' should be provided, but not a mixture.

Parameters:

topicPartitions - to set.

Since:

2.3

See Also:

• setTopics(String...)
• setTopicPattern(Pattern)

getTopicPartitionsToAssign

@Nullable
public TopicPartitionOffset[] getTopicPartitionsToAssign()

Return the topicPartitions for this endpoint.

Specified by:

getTopicPartitionsToAssign in interface KafkaListenerEndpoint

Returns:

the topicPartitions for this endpoint.

Since:

2.3

setTopicPattern

public void setTopicPattern(Pattern topicPattern)

Set the topic pattern to use. Cannot be used with topics or topicPartitions.

Parameters:

topicPattern - the pattern

See Also:

• setTopicPartitions(TopicPartitionOffset...)
• setTopics(String...)

getTopicPattern

@Nullable
public Pattern getTopicPattern()

Return the topicPattern for this endpoint.

Specified by:

getTopicPattern in interface KafkaListenerEndpoint

Returns:

the topicPattern for this endpoint.

getGroup

@Nullable
public String getGroup()

Description copied from interface: KafkaListenerEndpoint

Return the group of this endpoint or null if not in a group.

Specified by:

getGroup in interface KafkaListenerEndpoint

Returns:

the group of this endpoint or null if not in a group.

setGroup

public void setGroup(String group)

Set the group for the corresponding listener container.

Parameters:

group - the group.

isBatchListener

public boolean isBatchListener()

Return true if this endpoint creates a batch listener.

Returns:

true for a batch listener.

Since:

1.1

getBatchListener

@Nullable
public Boolean getBatchListener()

Return the current batch listener flag for this endpoint, or null if not explicitly set.

Specified by:

getBatchListener in interface KafkaListenerEndpoint

Returns:

the batch listener flag.

Since:

2.8

setBatchListener

public void setBatchListener(boolean batchListener)

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

Parameters:

batchListener - true for a batch listener.

Since:

1.1

setReplyTemplate

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

Set the KafkaTemplate to use to send replies.

Parameters:

replyTemplate - the template.

Since:

2.0

getReplyTemplate

@Nullable
protected KafkaTemplate<?,?> getReplyTemplate()

getRecordFilterStrategy

@Nullable
protected RecordFilterStrategy<? super K,? super V> getRecordFilterStrategy()

setRecordFilterStrategy

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

Set a RecordFilterStrategy implementation.

Parameters:

recordFilterStrategy - the strategy implementation.

isAckDiscarded

protected boolean isAckDiscarded()

setAckDiscarded

public void setAckDiscarded(boolean ackDiscarded)

Set to true if the setRecordFilterStrategy(RecordFilterStrategy) is in use.

Parameters:

ackDiscarded - the ackDiscarded.

getClientIdPrefix

@Nullable
public String getClientIdPrefix()

Description copied from interface: KafkaListenerEndpoint

Return the client id prefix for the container; it will be suffixed by '-n' to provide a unique id when concurrency is used.

Specified by:

getClientIdPrefix in interface KafkaListenerEndpoint

Returns:

the client id prefix.

setClientIdPrefix

public void setClientIdPrefix(String clientIdPrefix)

Set the client id prefix; overrides the client id in the consumer configuration properties.

Parameters:

clientIdPrefix - the prefix.

Since:

2.1.1

getConcurrency

@Nullable
public Integer getConcurrency()

Description copied from interface: KafkaListenerEndpoint

Return the concurrency for this endpoint's container.

Specified by:

getConcurrency in interface KafkaListenerEndpoint

Returns:

the concurrency.

setConcurrency

public void setConcurrency(Integer concurrency)

Set the concurrency for this endpoint's container.

Parameters:

concurrency - the concurrency.

Since:

2.2

getAutoStartup

@Nullable
public Boolean getAutoStartup()

Description copied from interface: KafkaListenerEndpoint

Return the autoStartup for this endpoint's container.

Specified by:

getAutoStartup in interface KafkaListenerEndpoint

Returns:

the autoStartup.

setAutoStartup

public void setAutoStartup(Boolean autoStartup)

Set the autoStartup for this endpoint's container.

Parameters:

autoStartup - the autoStartup.

Since:

2.2

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

getConsumerProperties

@Nullable
public Properties getConsumerProperties()

Description copied from interface: KafkaListenerEndpoint

Get the consumer properties that will be merged with the consumer properties provided by the consumer factory; properties here will supersede any with the same name(s) in the consumer factory. group.id and client.id are ignored.

Specified by:

getConsumerProperties in interface KafkaListenerEndpoint

Returns:

the properties.

See Also:

• ConsumerConfig
• KafkaListenerEndpoint.getGroupId()
• KafkaListenerEndpoint.getClientIdPrefix()

setConsumerProperties

public void setConsumerProperties(Properties consumerProperties)

Set the consumer properties that will be merged with the consumer properties provided by the consumer factory; properties here will supersede any with the same name(s) in the consumer factory. group.id and client.id are ignored.

Parameters:

consumerProperties - the properties.

Since:

2.1.4

See Also:

• ConsumerConfig
• setGroupId(String)
• setClientIdPrefix(String)

isSplitIterables

public boolean isSplitIterables()

Description copied from interface: KafkaListenerEndpoint

When true, Iterable return results will be split into discrete records.

Specified by:

isSplitIterables in interface KafkaListenerEndpoint

Returns:

true to split.

setSplitIterables

public void setSplitIterables(boolean splitIterables)

Set to false to disable splitting Iterable reply values into separate records.

Parameters:

splitIterables - false to disable; default true.

Since:

2.3.5

getListenerInfo

@Nullable
public byte[] getListenerInfo()

Description copied from interface: KafkaListenerEndpoint

Get the listener info to insert in the record header.

Specified by:

getListenerInfo in interface KafkaListenerEndpoint

Returns:

the info.

setListenerInfo

public void setListenerInfo(@Nullable
 byte[] listenerInfo)

Set the listener info to insert in the record header.

Parameters:

listenerInfo - the info.

Since:

2.8.4

getBatchToRecordAdapter

@Nullable
protected BatchToRecordAdapter<K,V> getBatchToRecordAdapter()

setBatchToRecordAdapter

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

Set a BatchToRecordAdapter.

Parameters:

batchToRecordAdapter - the adapter.

Since:

2.4.2

setCorrelationHeaderName

public void setCorrelationHeaderName(String correlationHeaderName)

Set a custom header name for the correlation id. Default KafkaHeaders.CORRELATION_ID. This header will be echoed back in any reply message.

Parameters:

correlationHeaderName - the header name.

Since:

3.0

afterPropertiesSet

public void afterPropertiesSet()

Specified by:

afterPropertiesSet in interface InitializingBean

setupListenerContainer

public void setupListenerContainer(MessageListenerContainer listenerContainer,
 @Nullable
 MessageConverter messageConverter)

Description copied from interface: KafkaListenerEndpoint

Setup the specified message listener container with the model defined by this endpoint.

This endpoint must provide the requested missing option(s) of the specified container to make it usable. Usually, this is about setting the queues and the messageListener to use but an implementation may override any default setting that was already set.

Specified by:

setupListenerContainer in interface KafkaListenerEndpoint

Parameters:

listenerContainer - the listener container to configure

messageConverter - the message converter - can be null

createMessageListener

protected abstract MessagingMessageListenerAdapter<K,V> createMessageListener(MessageListenerContainer container,
 @Nullable
 MessageConverter messageConverter)

Create a MessageListener that is able to serve this endpoint for the specified container.

Parameters:

container - the MessageListenerContainer to create a MessageListener.

messageConverter - the message converter - may be null.

Returns:

a MessageListener instance.

getEndpointDescription

protected StringBuilder getEndpointDescription()

Return a description for this endpoint.

Returns:

a description for this endpoint.

Available to subclasses, for inclusion in their toString() result.

toString

public String toString()

Overrides:

toString in class Object