K
- the key type.V
- the value type.public class KafkaMessageSource<K,V>
extends org.springframework.integration.endpoint.AbstractMessageSource<java.lang.Object>
implements org.springframework.integration.core.Pausable
NOTE: If the application acknowledges messages out of order, the acks will be deferred until all messages prior to the offset are ack'd. If multiple records are retrieved and an earlier offset is requeued, records from the subsequent offsets will be redelivered - even if they were processed successfully. Applications should therefore implement idempotency.
Starting with version 3.1.2, this source implements Pausable
which
allows you to pause and resume the Consumer
. While the consumer is
paused, you must continue to call AbstractMessageSource.receive()
within
max.poll.interval.ms
, to prevent a rebalance.
Modifier and Type | Class and Description |
---|---|
static class |
KafkaMessageSource.KafkaAckCallback<K,V>
AcknowledgmentCallback for Kafka.
|
static class |
KafkaMessageSource.KafkaAckCallbackFactory<K,V>
AcknowledgmentCallbackFactory for KafkaAckInfo.
|
static interface |
KafkaMessageSource.KafkaAckInfo<K,V>
Information for building an KafkaAckCallback.
|
class |
KafkaMessageSource.KafkaAckInfoImpl
Information for building an KafkaAckCallback.
|
Modifier and Type | Field and Description |
---|---|
static java.lang.String |
REMAINING_RECORDS
The number of records remaining from the previous poll.
|
Constructor and Description |
---|
KafkaMessageSource(org.springframework.kafka.core.ConsumerFactory<K,V> consumerFactory,
org.springframework.kafka.listener.ConsumerProperties consumerProperties)
Construct an instance with the supplied parameters.
|
KafkaMessageSource(org.springframework.kafka.core.ConsumerFactory<K,V> consumerFactory,
org.springframework.kafka.listener.ConsumerProperties consumerProperties,
boolean allowMultiFetch)
Construct an instance with the supplied parameters.
|
KafkaMessageSource(org.springframework.kafka.core.ConsumerFactory<K,V> consumerFactory,
org.springframework.kafka.listener.ConsumerProperties consumerProperties,
KafkaMessageSource.KafkaAckCallbackFactory<K,V> ackCallbackFactory)
Construct an instance with the supplied parameters.
|
KafkaMessageSource(org.springframework.kafka.core.ConsumerFactory<K,V> consumerFactory,
org.springframework.kafka.listener.ConsumerProperties consumerProperties,
KafkaMessageSource.KafkaAckCallbackFactory<K,V> ackCallbackFactory,
boolean allowMultiFetch)
Construct an instance with the supplied parameters.
|
KafkaMessageSource(org.springframework.kafka.core.ConsumerFactory<K,V> consumerFactory,
KafkaMessageSource.KafkaAckCallbackFactory<K,V> ackCallbackFactory,
java.lang.String... topics)
Deprecated.
|
KafkaMessageSource(org.springframework.kafka.core.ConsumerFactory<K,V> consumerFactory,
java.lang.String... topics)
Deprecated.
|
Modifier and Type | Method and Description |
---|---|
protected void |
createConsumer() |
void |
destroy() |
protected java.lang.Object |
doReceive() |
protected java.lang.String |
getClientId() |
protected java.time.Duration |
getCommitTimeout() |
java.lang.String |
getComponentType() |
protected java.lang.String |
getGroupId() |
protected org.springframework.kafka.support.converter.RecordMessageConverter |
getMessageConverter() |
protected java.lang.Class<?> |
getPayloadType() |
protected long |
getPollTimeout() |
protected org.apache.kafka.clients.consumer.ConsumerRebalanceListener |
getRebalanceListener() |
protected boolean |
isRawMessageHeader() |
boolean |
isRunning() |
protected void |
onInit() |
void |
pause() |
void |
resume() |
void |
setClientId(java.lang.String clientId)
Deprecated.
in favor of using
ConsumerProperties |
void |
setGroupId(java.lang.String groupId)
Deprecated.
in favor of using
ConsumerProperties |
void |
setMessageConverter(org.springframework.kafka.support.converter.RecordMessageConverter messageConverter)
Set the message converter to replace the default
MessagingMessageConverter . |
void |
setPayloadType(java.lang.Class<?> payloadType)
Set the payload type.
|
void |
setPollTimeout(long pollTimeout)
Deprecated.
in favor of using
ConsumerProperties |
void |
setRawMessageHeader(boolean rawMessageHeader)
Set to true to include the raw
ConsumerRecord as headers with keys
KafkaHeaders.RAW_DATA and
IntegrationMessageHeaderAccessor.SOURCE_DATA . |
void |
setRebalanceListener(org.apache.kafka.clients.consumer.ConsumerRebalanceListener rebalanceListener)
Deprecated.
in favor of using
ConsumerProperties |
void |
start() |
void |
stop() |
buildMessage, getBeanName, getComponentName, getManagedName, getManagedType, getMessageCount, getMessageCountLong, getOverrides, isCountsEnabled, isLoggingEnabled, receive, registerMetricsCaptor, reset, setBeanName, setCountsEnabled, setHeaderExpressions, setLoggingEnabled, setManagedName, setManagedType
afterPropertiesSet, evaluateExpression, evaluateExpression, evaluateExpression, evaluateExpression, evaluateExpression, evaluateExpression, evaluateExpression, getBeanFactory, getEvaluationContext, getEvaluationContext, getMessageBuilderFactory, setBeanFactory, setConversionService
public static final java.lang.String REMAINING_RECORDS
@Deprecated public KafkaMessageSource(org.springframework.kafka.core.ConsumerFactory<K,V> consumerFactory, java.lang.String... topics)
KafkaMessageSource(ConsumerFactory, ConsumerProperties)
consumerFactory
- the consumer factory.topics
- the topics.KafkaMessageSource(ConsumerFactory, ConsumerProperties, KafkaAckCallbackFactory, boolean)
@Deprecated public KafkaMessageSource(org.springframework.kafka.core.ConsumerFactory<K,V> consumerFactory, KafkaMessageSource.KafkaAckCallbackFactory<K,V> ackCallbackFactory, java.lang.String... topics)
KafkaMessageSource(ConsumerFactory, ConsumerProperties, KafkaAckCallbackFactory)
consumerFactory
- the consumer factory.ackCallbackFactory
- the ack callback factory.topics
- the topics.KafkaMessageSource(ConsumerFactory, ConsumerProperties, KafkaAckCallbackFactory, boolean)
public KafkaMessageSource(org.springframework.kafka.core.ConsumerFactory<K,V> consumerFactory, org.springframework.kafka.listener.ConsumerProperties consumerProperties)
consumerFactory
- the consumer factory.consumerProperties
- the consumer properties.KafkaMessageSource(ConsumerFactory, ConsumerProperties, KafkaAckCallbackFactory, boolean)
public KafkaMessageSource(org.springframework.kafka.core.ConsumerFactory<K,V> consumerFactory, org.springframework.kafka.listener.ConsumerProperties consumerProperties, boolean allowMultiFetch)
max.poll.records
to be fetched on each poll. When false
(default) max.poll.records
is coerced to 1 if the consumer factory is a
DefaultKafkaConsumerFactory
or otherwise rejected with an
IllegalArgumentException
. IMPORTANT: When true, you must call
AbstractMessageSource.receive()
at a sufficient rate to consume the number of records received
within max.poll.interval.ms
. When false, you must call AbstractMessageSource.receive()
within max.poll.interval.ms
. pause()
will not take effect until
the records from the previous poll are consumed.consumerFactory
- the consumer factory.consumerProperties
- the consumer properties.allowMultiFetch
- true to allow max.poll.records > 1
.public KafkaMessageSource(org.springframework.kafka.core.ConsumerFactory<K,V> consumerFactory, org.springframework.kafka.listener.ConsumerProperties consumerProperties, KafkaMessageSource.KafkaAckCallbackFactory<K,V> ackCallbackFactory)
consumerFactory
- the consumer factory.consumerProperties
- the consumer properties.ackCallbackFactory
- the ack callback factory.KafkaMessageSource(ConsumerFactory, ConsumerProperties, KafkaAckCallbackFactory, boolean)
public KafkaMessageSource(org.springframework.kafka.core.ConsumerFactory<K,V> consumerFactory, org.springframework.kafka.listener.ConsumerProperties consumerProperties, KafkaMessageSource.KafkaAckCallbackFactory<K,V> ackCallbackFactory, boolean allowMultiFetch)
max.poll.records
to be fetched on each poll. When false
(default) max.poll.records
is coerced to 1 if the consumer factory is a
DefaultKafkaConsumerFactory
or otherwise rejected with an
IllegalArgumentException
. IMPORTANT: When true, you must call
AbstractMessageSource.receive()
at a sufficient rate to consume the number of records received
within max.poll.interval.ms
. When false, you must call AbstractMessageSource.receive()
within max.poll.interval.ms
. pause()
will not take effect until
the records from the previous poll are consumed.consumerFactory
- the consumer factory.consumerProperties
- the consumer properties.ackCallbackFactory
- the ack callback factory.allowMultiFetch
- true to allow max.poll.records > 1
.protected void onInit()
onInit
in class org.springframework.integration.util.AbstractExpressionEvaluator
protected java.lang.String getGroupId()
@Deprecated public void setGroupId(java.lang.String groupId)
ConsumerProperties
groupId
- the group id.ConsumerProperties
protected java.lang.String getClientId()
@Deprecated public void setClientId(java.lang.String clientId)
ConsumerProperties
clientId
- the client id.ConsumerProperties
protected long getPollTimeout()
@Deprecated public void setPollTimeout(long pollTimeout)
ConsumerProperties
pollTimeout
- the poll timeout.ConsumerProperties
protected org.springframework.kafka.support.converter.RecordMessageConverter getMessageConverter()
public void setMessageConverter(org.springframework.kafka.support.converter.RecordMessageConverter messageConverter)
MessagingMessageConverter
.messageConverter
- the converter.protected java.lang.Class<?> getPayloadType()
public void setPayloadType(java.lang.Class<?> payloadType)
payloadType
- the type to convert to.protected org.apache.kafka.clients.consumer.ConsumerRebalanceListener getRebalanceListener()
@Deprecated public void setRebalanceListener(org.apache.kafka.clients.consumer.ConsumerRebalanceListener rebalanceListener)
ConsumerProperties
rebalanceListener
- the rebalance listener.ConsumerProperties
public java.lang.String getComponentType()
getComponentType
in interface org.springframework.integration.support.context.NamedComponent
protected boolean isRawMessageHeader()
public void setRawMessageHeader(boolean rawMessageHeader)
ConsumerRecord
as headers with keys
KafkaHeaders.RAW_DATA
and
IntegrationMessageHeaderAccessor.SOURCE_DATA
. enabling callers to have
access to the record to process errors.rawMessageHeader
- true to include the header.protected java.time.Duration getCommitTimeout()
public boolean isRunning()
isRunning
in interface org.springframework.context.Lifecycle
public void start()
start
in interface org.springframework.context.Lifecycle
public void stop()
stop
in interface org.springframework.context.Lifecycle
public void pause()
pause
in interface org.springframework.integration.core.Pausable
public void resume()
resume
in interface org.springframework.integration.core.Pausable
protected java.lang.Object doReceive()
doReceive
in class org.springframework.integration.endpoint.AbstractMessageSource<java.lang.Object>
protected void createConsumer()
public void destroy()
destroy
in interface org.springframework.beans.factory.DisposableBean
destroy
in interface org.springframework.integration.support.management.IntegrationManagement
destroy
in class org.springframework.integration.endpoint.AbstractMessageSource<java.lang.Object>