org.springframework.integration.kafka.inbound

Class KafkaMessageDrivenChannelAdapter<K,V>

java.lang.Object

org.springframework.integration.context.IntegrationObjectSupport

org.springframework.integration.endpoint.AbstractEndpoint

org.springframework.integration.endpoint.MessageProducerSupport

org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter<K,V>

Type Parameters:

K - the key type.

V - the value type.

All Implemented Interfaces:

Aware, BeanFactoryAware, BeanNameAware, DisposableBean, InitializingBean, SmartInitializingSingleton, ApplicationContextAware, Lifecycle, Phased, SmartLifecycle, ExpressionCapable, OrderlyShutdownCapable, MessageProducer, Pausable, IntegrationPattern, NamedComponent, ManageableLifecycle, ManageableSmartLifecycle, TrackableComponent

public class KafkaMessageDrivenChannelAdapter<K,V>
extends MessageProducerSupport
implements OrderlyShutdownCapable, Pausable

Message-driven channel adapter.

Since:

5.4

Author:

Marius Bogoevici, Gary Russell, Artem Bilan, Urs Keller

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	KafkaMessageDrivenChannelAdapter.ListenerMode The listener mode for the container, record or batch.

Field Summary

Fields inherited from class org.springframework.integration.endpoint.AbstractEndpoint

lifecycleCondition, lifecycleLock

Fields inherited from class org.springframework.integration.context.IntegrationObjectSupport

EXPRESSION_PARSER, logger

Fields inherited from interface org.springframework.context.SmartLifecycle

DEFAULT_PHASE

Constructor Summary

Constructors

Constructor and Description
KafkaMessageDrivenChannelAdapter(org.springframework.kafka.listener.AbstractMessageListenerContainer<K,V> messageListenerContainer) Construct an instance with mode KafkaMessageDrivenChannelAdapter.ListenerMode.record.
KafkaMessageDrivenChannelAdapter(org.springframework.kafka.listener.AbstractMessageListenerContainer<K,V> messageListenerContainer, KafkaMessageDrivenChannelAdapter.ListenerMode mode) Construct an instance with the provided mode.

Method Summary

Modifier and Type	Method and Description
int	afterShutdown() Called after normal shutdown of schedulers, executors etc, and after the shutdown delay has elapsed, but before any forced shutdown of any remaining active scheduler/executor threads.Can optionally return the number of active messages still in process.
int	beforeShutdown() Called before shutdown begins.
protected void	doStart() Take no action by default.
protected void	doStop() Take no action by default.
String	getComponentType() Subclasses may implement this method to provide component type information.
protected AttributeAccessor	getErrorMessageAttributes(Message<?> message) Populate an AttributeAccessor to be used when building an error message with the errorMessageStrategy.
boolean	isPaused() Check if the endpoint is paused.
protected void	onInit() Subclasses may implement this for initialization logic.
void	pause() Pause the endpoint.
void	resume() Resume the endpoint if paused.
void	setAckDiscarded(boolean ackDiscarded) A boolean flag to indicate if FilteringMessageListenerAdapter should acknowledge discarded records or not.
void	setBatchMessageConverter(org.springframework.kafka.support.converter.BatchMessageConverter messageConverter) Set the message converter to use with a batch-based consumer.
void	setBindSourceRecord(boolean bindSourceRecord) Set to true to bind the source consumer record in the header named IntegrationMessageHeaderAccessor.SOURCE_DATA.
void	setFilterInRetry(boolean filterInRetry) The boolean flag to specify the order how RetryingMessageListenerAdapter and FilteringMessageListenerAdapter are wrapped to each other, if both of them are present.
void	setMessageConverter(org.springframework.kafka.support.converter.MessageConverter messageConverter) Set the message converter; must be a RecordMessageConverter or BatchMessageConverter depending on mode.
void	setOnPartitionsAssignedSeekCallback(java.util.function.BiConsumer<Map<org.apache.kafka.common.TopicPartition,Long>,org.springframework.kafka.listener.ConsumerSeekAware.ConsumerSeekCallback> onPartitionsAssignedCallback) Specify a BiConsumer for seeks management during ConsumerSeekAware.ConsumerSeekCallback#onPartitionsAssigned(Map, ConsumerSeekAware.ConsumerSeekCallback) call from the KafkaMessageListenerContainer.
void	setPayloadType(Class<?> payloadType) When using a type-aware message converter (such as StringJsonMessageConverter, set the payload type the converter should create.
void	setRecordFilterStrategy(org.springframework.kafka.listener.adapter.RecordFilterStrategy<K,V> recordFilterStrategy) Specify a RecordFilterStrategy to wrap KafkaMessageDrivenChannelAdapter.IntegrationRecordMessageListener into FilteringMessageListenerAdapter.
void	setRecordMessageConverter(org.springframework.kafka.support.converter.RecordMessageConverter messageConverter) Set the message converter to use with a record-based consumer.
void	setRecoveryCallback(org.springframework.retry.RecoveryCallback<? extends Object> recoveryCallback) A RecoveryCallback instance for retry operation; if null, the exception will be thrown to the container after retries are exhausted (unless an error channel is configured).
void	setRetryTemplate(org.springframework.retry.support.RetryTemplate retryTemplate) Specify a RetryTemplate instance to wrap KafkaMessageDrivenChannelAdapter.IntegrationRecordMessageListener into RetryingMessageListenerAdapter.

Methods inherited from class org.springframework.integration.endpoint.MessageProducerSupport

afterSingletonsInstantiated, buildErrorMessage, getErrorChannel, getIntegrationPatternType, getMessagingTemplate, getOutputChannel, sendErrorMessageIfNecessary, sendMessage, setErrorChannel, setErrorChannelName, setErrorMessageStrategy, setOutputChannel, setOutputChannelName, setSendTimeout, setShouldTrack, subscribeToPublisher

Methods inherited from class org.springframework.integration.endpoint.AbstractEndpoint

destroy, doStop, getPhase, getRole, isActive, isAutoStartup, isRunning, setAutoStartup, setPhase, setRole, start, stop, stop

Methods inherited from class org.springframework.integration.context.IntegrationObjectSupport

afterPropertiesSet, extractTypeIfPossible, generateId, getApplicationContext, getApplicationContextId, getBeanDescription, getBeanFactory, getBeanName, getChannelResolver, getComponentName, getConversionService, getExpression, getIntegrationProperties, getIntegrationProperty, getMessageBuilderFactory, getTaskScheduler, isInitialized, setApplicationContext, setBeanFactory, setBeanName, setChannelResolver, setComponentName, setConversionService, setMessageBuilderFactory, setPrimaryExpression, setTaskScheduler, toString

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.springframework.integration.support.management.ManageableLifecycle

isRunning, start, stop

Methods inherited from interface org.springframework.integration.support.context.NamedComponent

getBeanName, getComponentName

Constructor Detail

KafkaMessageDrivenChannelAdapter

public KafkaMessageDrivenChannelAdapter(org.springframework.kafka.listener.AbstractMessageListenerContainer<K,V> messageListenerContainer)

Construct an instance with mode KafkaMessageDrivenChannelAdapter.ListenerMode.record.

Parameters:

messageListenerContainer - the container.

KafkaMessageDrivenChannelAdapter

public KafkaMessageDrivenChannelAdapter(org.springframework.kafka.listener.AbstractMessageListenerContainer<K,V> messageListenerContainer,
                                        KafkaMessageDrivenChannelAdapter.ListenerMode mode)

Construct an instance with the provided mode.

Parameters:

messageListenerContainer - the container.

mode - the mode.

Since:

1.2

Method Detail

setMessageConverter

public void setMessageConverter(org.springframework.kafka.support.converter.MessageConverter messageConverter)

Set the message converter; must be a RecordMessageConverter or BatchMessageConverter depending on mode.

Parameters:

messageConverter - the converter.

setRecordMessageConverter

public void setRecordMessageConverter(org.springframework.kafka.support.converter.RecordMessageConverter messageConverter)

Set the message converter to use with a record-based consumer.

Parameters:

messageConverter - the converter.

Since:

2.1

setBatchMessageConverter

public void setBatchMessageConverter(org.springframework.kafka.support.converter.BatchMessageConverter messageConverter)

Set the message converter to use with a batch-based consumer.

Parameters:

messageConverter - the converter.

Since:

2.1

setRecordFilterStrategy

public void setRecordFilterStrategy(org.springframework.kafka.listener.adapter.RecordFilterStrategy<K,V> recordFilterStrategy)

Specify a RecordFilterStrategy to wrap KafkaMessageDrivenChannelAdapter.IntegrationRecordMessageListener into FilteringMessageListenerAdapter.

Parameters:

recordFilterStrategy - the RecordFilterStrategy to use.

Since:

2.0.1

setAckDiscarded

public void setAckDiscarded(boolean ackDiscarded)

A boolean flag to indicate if FilteringMessageListenerAdapter should acknowledge discarded records or not. Does not make sense if setRecordFilterStrategy(RecordFilterStrategy) isn't specified.

Parameters:

ackDiscarded - true to ack (commit offset for) discarded messages.

Since:

2.0.1

setRetryTemplate

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

Specify a RetryTemplate instance to wrap KafkaMessageDrivenChannelAdapter.IntegrationRecordMessageListener into RetryingMessageListenerAdapter.

Parameters:

retryTemplate - the RetryTemplate to use.

Since:

2.0.1

setRecoveryCallback

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

A RecoveryCallback instance for retry operation; if null, the exception will be thrown to the container after retries are exhausted (unless an error channel is configured). Does not make sense if setRetryTemplate(RetryTemplate) isn't specified.

Parameters:

recoveryCallback - the recovery callback.

Since:

2.0.1

setFilterInRetry

public void setFilterInRetry(boolean filterInRetry)

The boolean flag to specify the order how RetryingMessageListenerAdapter and FilteringMessageListenerAdapter are wrapped to each other, if both of them are present. Does not make sense if only one of RetryTemplate or RecordFilterStrategy is present, or any.

Parameters:

filterInRetry - the order for RetryingMessageListenerAdapter and FilteringMessageListenerAdapter wrapping. Defaults to false.

Since:

2.0.1

setPayloadType

public void setPayloadType(Class<?> payloadType)

When using a type-aware message converter (such as StringJsonMessageConverter, set the payload type the converter should create. Defaults to Object.

Parameters:

payloadType - the type.

Since:

2.1.1

setOnPartitionsAssignedSeekCallback

public void setOnPartitionsAssignedSeekCallback(java.util.function.BiConsumer<Map<org.apache.kafka.common.TopicPartition,Long>,org.springframework.kafka.listener.ConsumerSeekAware.ConsumerSeekCallback> onPartitionsAssignedCallback)

Specify a BiConsumer for seeks management during ConsumerSeekAware.ConsumerSeekCallback#onPartitionsAssigned(Map, ConsumerSeekAware.ConsumerSeekCallback) call from the KafkaMessageListenerContainer. This is called from the internal MessagingMessageListenerAdapter implementation.

Parameters:

onPartitionsAssignedCallback - the BiConsumer to use

Since:

3.0.4

See Also:

ConsumerSeekAware.onPartitionsAssigned(java.util.Map<org.apache.kafka.common.TopicPartition, java.lang.Long>, org.springframework.kafka.listener.ConsumerSeekAware.ConsumerSeekCallback)

setBindSourceRecord

public void setBindSourceRecord(boolean bindSourceRecord)

Set to true to bind the source consumer record in the header named IntegrationMessageHeaderAccessor.SOURCE_DATA. Does not apply to batch listeners.

Parameters:

bindSourceRecord - true to bind.

Since:

3.1.4

getComponentType

public String getComponentType()

Description copied from class: IntegrationObjectSupport

Subclasses may implement this method to provide component type information.

Specified by:

getComponentType in interface NamedComponent

Overrides:

getComponentType in class IntegrationObjectSupport

onInit

protected void onInit()

Description copied from class: IntegrationObjectSupport

Subclasses may implement this for initialization logic.

Overrides:

onInit in class MessageProducerSupport

doStart

protected void doStart()

Description copied from class: MessageProducerSupport

Take no action by default. Subclasses may override this if they need lifecycle-managed behavior. Protected by 'lifecycleLock'.

Overrides:

doStart in class MessageProducerSupport

doStop

protected void doStop()

Description copied from class: MessageProducerSupport

Take no action by default. Subclasses may override this if they need lifecycle-managed behavior.

Overrides:

doStop in class MessageProducerSupport

pause

public void pause()

Description copied from interface: Pausable

Pause the endpoint.

Specified by:

pause in interface Pausable

resume

public void resume()

Description copied from interface: Pausable

Resume the endpoint if paused.

Specified by:

resume in interface Pausable

isPaused

public boolean isPaused()

Description copied from interface: Pausable

Check if the endpoint is paused.

Specified by:

isPaused in interface Pausable

Returns:

true if paused.

beforeShutdown

public int beforeShutdown()

Description copied from interface: OrderlyShutdownCapable

Called before shutdown begins. Implementations should stop accepting new messages. Can optionally return the number of active messages in process.

Specified by:

beforeShutdown in interface OrderlyShutdownCapable

Returns:

The number of active messages if available.

afterShutdown

public int afterShutdown()

Description copied from interface: OrderlyShutdownCapable

Called after normal shutdown of schedulers, executors etc, and after the shutdown delay has elapsed, but before any forced shutdown of any remaining active scheduler/executor threads.Can optionally return the number of active messages still in process.

Specified by:

afterShutdown in interface OrderlyShutdownCapable

Returns:

The number of active messages if available.

getErrorMessageAttributes

protected AttributeAccessor getErrorMessageAttributes(Message<?> message)

Description copied from class: MessageProducerSupport

Populate an AttributeAccessor to be used when building an error message with the errorMessageStrategy.

Overrides:

getErrorMessageAttributes in class MessageProducerSupport

Parameters:

message - the message.

Returns:

the attributes.