K
- the key type.V
- the value type.public class KafkaMessageDrivenChannelAdapter<K,V>
extends org.springframework.integration.endpoint.MessageProducerSupport
implements org.springframework.integration.context.OrderlyShutdownCapable, org.springframework.integration.core.Pausable
Modifier and Type | Class and Description |
---|---|
static class |
KafkaMessageDrivenChannelAdapter.ListenerMode
The listener mode for the container, record or batch.
|
lifecycleCondition, lifecycleLock
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.
|
Modifier and Type | Method and Description |
---|---|
int |
afterShutdown() |
int |
beforeShutdown() |
protected void |
doStart() |
protected void |
doStop() |
java.lang.String |
getComponentType() |
protected org.springframework.core.AttributeAccessor |
getErrorMessageAttributes(org.springframework.messaging.Message<?> message) |
protected void |
onInit() |
void |
pause() |
void |
resume() |
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<java.util.Map<org.apache.kafka.common.TopicPartition,java.lang.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(java.lang.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 java.lang.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 . |
afterSingletonsInstantiated, buildErrorMessage, getErrorChannel, getMessagingTemplate, getOutputChannel, sendErrorMessageIfNecessary, sendMessage, setErrorChannel, setErrorChannelName, setErrorMessageStrategy, setOutputChannel, setOutputChannelName, setSendTimeout, setShouldTrack
destroy, doStop, getPhase, getRole, isAutoStartup, isRunning, setAutoStartup, setPhase, setRole, start, stop, stop
afterPropertiesSet, extractTypeIfPossible, generateId, getApplicationContext, getApplicationContextId, getBeanFactory, getBeanName, getChannelResolver, getComponentName, getConversionService, getExpression, getIntegrationProperties, getIntegrationProperty, getMessageBuilderFactory, getTaskScheduler, isInitialized, setApplicationContext, setBeanFactory, setBeanName, setChannelResolver, setComponentName, setConversionService, setMessageBuilderFactory, setPrimaryExpression, setTaskScheduler, toString
public KafkaMessageDrivenChannelAdapter(org.springframework.kafka.listener.AbstractMessageListenerContainer<K,V> messageListenerContainer)
KafkaMessageDrivenChannelAdapter.ListenerMode.record
.messageListenerContainer
- the container.public KafkaMessageDrivenChannelAdapter(org.springframework.kafka.listener.AbstractMessageListenerContainer<K,V> messageListenerContainer, KafkaMessageDrivenChannelAdapter.ListenerMode mode)
messageListenerContainer
- the container.mode
- the mode.public void setMessageConverter(org.springframework.kafka.support.converter.MessageConverter messageConverter)
RecordMessageConverter
or
BatchMessageConverter
depending on mode.messageConverter
- the converter.public void setRecordMessageConverter(org.springframework.kafka.support.converter.RecordMessageConverter messageConverter)
messageConverter
- the converter.public void setBatchMessageConverter(org.springframework.kafka.support.converter.BatchMessageConverter messageConverter)
messageConverter
- the converter.public void setRecordFilterStrategy(org.springframework.kafka.listener.adapter.RecordFilterStrategy<K,V> recordFilterStrategy)
RecordFilterStrategy
to wrap
KafkaMessageDrivenChannelAdapter.IntegrationRecordMessageListener
into
FilteringMessageListenerAdapter
.recordFilterStrategy
- the RecordFilterStrategy
to use.public void setAckDiscarded(boolean ackDiscarded)
boolean
flag to indicate if FilteringMessageListenerAdapter
should acknowledge discarded records or not.
Does not make sense if setRecordFilterStrategy(RecordFilterStrategy)
isn't specified.ackDiscarded
- true to ack (commit offset for) discarded messages.public void setRetryTemplate(org.springframework.retry.support.RetryTemplate retryTemplate)
RetryTemplate
instance to wrap
KafkaMessageDrivenChannelAdapter.IntegrationRecordMessageListener
into
RetryingMessageListenerAdapter
.retryTemplate
- the RetryTemplate
to use.public void setRecoveryCallback(org.springframework.retry.RecoveryCallback<? extends java.lang.Object> recoveryCallback)
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.recoveryCallback
- the recovery callback.public void setFilterInRetry(boolean filterInRetry)
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.filterInRetry
- the order for RetryingMessageListenerAdapter
and
FilteringMessageListenerAdapter
wrapping. Defaults to false
.public void setPayloadType(java.lang.Class<?> payloadType)
StringJsonMessageConverter
,
set the payload type the converter should create. Defaults to Object
.payloadType
- the type.public void setOnPartitionsAssignedSeekCallback(java.util.function.BiConsumer<java.util.Map<org.apache.kafka.common.TopicPartition,java.lang.Long>,org.springframework.kafka.listener.ConsumerSeekAware.ConsumerSeekCallback> onPartitionsAssignedCallback)
BiConsumer
for seeks management during
ConsumerSeekAware.ConsumerSeekCallback#onPartitionsAssigned(Map, ConsumerSeekAware.ConsumerSeekCallback)
call from the KafkaMessageListenerContainer
.
This is called from the internal
MessagingMessageListenerAdapter
implementation.onPartitionsAssignedCallback
- the BiConsumer
to useConsumerSeekAware.onPartitionsAssigned(java.util.Map<org.apache.kafka.common.TopicPartition, java.lang.Long>, org.springframework.kafka.listener.ConsumerSeekAware.ConsumerSeekCallback)
public void setBindSourceRecord(boolean bindSourceRecord)
IntegrationMessageHeaderAccessor.SOURCE_DATA
.
Does not apply to batch listeners.bindSourceRecord
- true to bind.public java.lang.String getComponentType()
getComponentType
in interface org.springframework.integration.support.context.NamedComponent
getComponentType
in class org.springframework.integration.context.IntegrationObjectSupport
protected void onInit()
onInit
in class org.springframework.integration.endpoint.MessageProducerSupport
protected void doStart()
doStart
in class org.springframework.integration.endpoint.MessageProducerSupport
protected void doStop()
doStop
in class org.springframework.integration.endpoint.MessageProducerSupport
public void pause()
pause
in interface org.springframework.integration.core.Pausable
public void resume()
resume
in interface org.springframework.integration.core.Pausable
public int beforeShutdown()
beforeShutdown
in interface org.springframework.integration.context.OrderlyShutdownCapable
public int afterShutdown()
afterShutdown
in interface org.springframework.integration.context.OrderlyShutdownCapable
protected org.springframework.core.AttributeAccessor getErrorMessageAttributes(org.springframework.messaging.Message<?> message)
getErrorMessageAttributes
in class org.springframework.integration.endpoint.MessageProducerSupport