org.springframework.integration.dsl.kafka

Class KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V>

java.lang.Object

org.springframework.integration.dsl.kafka.KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V>

Type Parameters:

K - the key type.

V - the value type.

Enclosing class:

KafkaMessageDrivenChannelAdapterSpec<K,V,S extends KafkaMessageDrivenChannelAdapterSpec<K,V,S>>

public static class KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V>
extends java.lang.Object

A helper class in the Builder pattern style to delegate options to the ConcurrentMessageListenerContainer.

Method Summary

Modifier and Type	Method and Description
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V>	ackCount(int count) Set the number of outstanding record count after which offsets should be committed when AbstractMessageListenerContainer.AckMode#COUNT or AbstractMessageListenerContainer.AckMode#COUNT_TIME is being used.
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V>	ackMode(org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode ackMode) Set the ack mode to use when auto ack (in the configuration properties) is false.
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V>	ackOnError(boolean ackOnError) Set whether the container should ack messages that throw exceptions or not.
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V>	ackTime(long millis) Set the time (ms) after which outstanding offsets should be committed when AbstractMessageListenerContainer.AckMode#TIME or AbstractMessageListenerContainer.AckMode#COUNT_TIME is being used.
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V>	commitCallback(org.apache.kafka.clients.consumer.OffsetCommitCallback commitCallback) Set the commit callback; by default a simple logging callback is used to log success at DEBUG level and failures at ERROR level.
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V>	concurrency(int concurrency) Specify a concurrency maximum number for the AbstractMessageListenerContainer.
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V>	consumerRebalanceListener(org.apache.kafka.clients.consumer.ConsumerRebalanceListener consumerRebalanceListener) Set the user defined ConsumerRebalanceListener implementation.
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V>	consumerTaskExecutor(org.springframework.core.task.AsyncListenableTaskExecutor consumerTaskExecutor) Set the executor for threads that poll the consumer.
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V>	errorHandler(org.springframework.kafka.listener.ErrorHandler errorHandler) Specify an ErrorHandler for the AbstractMessageListenerContainer.
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V>	idleEventInterval(java.lang.Long idleEventInterval) Set the idle event interval; when set, an event is emitted if a poll returns no records and this interval has elapsed since a record was returned.
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V>	listenerTaskExecutor(org.springframework.core.task.AsyncListenableTaskExecutor listenerTaskExecutor) Set the executor for threads that invoke the listener.
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V>	pauseAfter(long pauseAfter) When using Kafka group management and pauseEnabled(boolean) is true, set the delay after which the consumer should be paused.
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V>	pauseEnabled(boolean pauseEnabled) Set to true to avoid rebalancing when this consumer is slow or throws a qualifying exception - pause the consumer.
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V>	pollTimeout(long pollTimeout) Set the max time to block in the consumer waiting for records.
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V>	queueDepth(int queueDepth) Set the queue depth for handoffs from the consumer thread to the listener thread.
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V>	shutdownTimeout(long shutdownTimeout) Set the timeout for shutting down the container.
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V>	syncCommits(boolean syncCommits) Set whether or not to call consumer.commitSync() or commitAsync() when the container is responsible for commits.

Methods inherited from class java.lang.Object

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

Method Detail

concurrency

public KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V> concurrency(int concurrency)

Specify a concurrency maximum number for the AbstractMessageListenerContainer.

Parameters:

concurrency - the concurrency maximum number.

Returns:

the spec.

See Also:

ConcurrentMessageListenerContainer.setConcurrency(int)

errorHandler

public KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V> errorHandler(org.springframework.kafka.listener.ErrorHandler errorHandler)

Specify an ErrorHandler for the AbstractMessageListenerContainer.

Parameters:

errorHandler - the ErrorHandler.

Returns:

the spec.

See Also:

ErrorHandler

ackMode

public KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V> ackMode(org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode ackMode)

Set the ack mode to use when auto ack (in the configuration properties) is false.

• RECORD: Ack after each record has been passed to the listener.
• BATCH: Ack after each batch of records received from the consumer has been passed to the listener
• TIME: Ack after this number of milliseconds; (should be greater than #setPollTimeout(long) pollTimeout.
• COUNT: Ack after at least this number of records have been received
• MANUAL: Listener is responsible for acking - use a AcknowledgingMessageListener.

Parameters:

ackMode - the AbstractMessageListenerContainer.AckMode; default BATCH.

Returns:

the spec.

See Also:

AbstractMessageListenerContainer.AckMode

pollTimeout

public KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V> pollTimeout(long pollTimeout)

Set the max time to block in the consumer waiting for records.

Parameters:

pollTimeout - the timeout in ms; default 1000.

Returns:

the spec.

See Also:

ContainerProperties.setPollTimeout(long)

ackCount

public KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V> ackCount(int count)

Set the number of outstanding record count after which offsets should be committed when AbstractMessageListenerContainer.AckMode#COUNT or AbstractMessageListenerContainer.AckMode#COUNT_TIME is being used.

Parameters:

count - the count

Returns:

the spec.

See Also:

ContainerProperties.setAckCount(int)

ackTime

public KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V> ackTime(long millis)

Set the time (ms) after which outstanding offsets should be committed when AbstractMessageListenerContainer.AckMode#TIME or AbstractMessageListenerContainer.AckMode#COUNT_TIME is being used. Should be larger than zero.

Parameters:

millis - the time

Returns:

the spec.

See Also:

ContainerProperties.setAckTime(long)

consumerTaskExecutor

public KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V> consumerTaskExecutor(org.springframework.core.task.AsyncListenableTaskExecutor consumerTaskExecutor)

Set the executor for threads that poll the consumer.

Parameters:

consumerTaskExecutor - the executor

Returns:

the spec.

See Also:

ContainerProperties.setConsumerTaskExecutor(AsyncListenableTaskExecutor)

listenerTaskExecutor

public KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V> listenerTaskExecutor(org.springframework.core.task.AsyncListenableTaskExecutor listenerTaskExecutor)

Set the executor for threads that invoke the listener.

Parameters:

listenerTaskExecutor - the executor

Returns:

the spec.

See Also:

ContainerProperties.setListenerTaskExecutor(AsyncListenableTaskExecutor)

pauseAfter

public KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V> pauseAfter(long pauseAfter)

When using Kafka group management and pauseEnabled(boolean) is true, set the delay after which the consumer should be paused. Default 10000.

Parameters:

pauseAfter - the delay.

Returns:

the spec.

See Also:

ContainerProperties.setPauseAfter(long)

pauseEnabled

public KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V> pauseEnabled(boolean pauseEnabled)

Set to true to avoid rebalancing when this consumer is slow or throws a qualifying exception - pause the consumer. Default: true.

Parameters:

pauseEnabled - true to pause.

Returns:

the spec.

See Also:

pauseAfter(long), ContainerProperties.setPauseEnabled(boolean)

queueDepth

public KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V> queueDepth(int queueDepth)

Set the queue depth for handoffs from the consumer thread to the listener thread. Default 1 (up to 2 in process).

Parameters:

queueDepth - the queue depth.

Returns:

the spec.

See Also:

ContainerProperties.setQueueDepth(int)

shutdownTimeout

public KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V> shutdownTimeout(long shutdownTimeout)

Set the timeout for shutting down the container. This is the maximum amount of time that the invocation to #stop(Runnable) will block for, before returning.

Parameters:

shutdownTimeout - the shutdown timeout.

Returns:

the spec.

See Also:

ContainerProperties.setShutdownTimeout(long)

consumerRebalanceListener

public KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V> consumerRebalanceListener(org.apache.kafka.clients.consumer.ConsumerRebalanceListener consumerRebalanceListener)

Set the user defined ConsumerRebalanceListener implementation.

Parameters:

consumerRebalanceListener - the ConsumerRebalanceListener instance

Returns:

the spec.

See Also:

ContainerProperties.setConsumerRebalanceListener(ConsumerRebalanceListener)

commitCallback

public KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V> commitCallback(org.apache.kafka.clients.consumer.OffsetCommitCallback commitCallback)

Set the commit callback; by default a simple logging callback is used to log success at DEBUG level and failures at ERROR level.

Parameters:

commitCallback - the callback.

Returns:

the spec.

See Also:

ContainerProperties.setCommitCallback(OffsetCommitCallback)

syncCommits

public KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V> syncCommits(boolean syncCommits)

Set whether or not to call consumer.commitSync() or commitAsync() when the container is responsible for commits. Default true. See https://github.com/spring-projects/spring-kafka/issues/62 At the time of writing, async commits are not entirely reliable.

Parameters:

syncCommits - true to use commitSync().

Returns:

the spec.

See Also:

ContainerProperties.setSyncCommits(boolean)

idleEventInterval

public KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V> idleEventInterval(java.lang.Long idleEventInterval)

Set the idle event interval; when set, an event is emitted if a poll returns no records and this interval has elapsed since a record was returned.

Parameters:

idleEventInterval - the interval.

Returns:

the spec.

See Also:

ContainerProperties.setIdleEventInterval(Long)

ackOnError

public KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K,V> ackOnError(boolean ackOnError)

Set whether the container should ack messages that throw exceptions or not.

Parameters:

ackOnError - whether the container should acknowledge messages that throw exceptions.

Returns:

the spec.

See Also:

ContainerProperties.setAckOnError(boolean)