Package org.springframework.kafka.listener

Class KafkaConsumerBackoffManager

java.lang.Object

org.springframework.kafka.listener.KafkaConsumerBackoffManager

All Implemented Interfaces:

java.util.EventListener, org.springframework.context.ApplicationListener<ListenerContainerPartitionIdleEvent>

public class KafkaConsumerBackoffManager
extends java.lang.Object
implements org.springframework.context.ApplicationListener<ListenerContainerPartitionIdleEvent>

A manager that backs off consumption for a given topic if the timestamp provided is not due. Use with SeekToCurrentErrorHandler to guarantee that the message is read again after partition consumption is resumed (or seek it manually by other means). It's also necessary to set a ContainerProperties.setIdlePartitionEventInterval(Long) so the Manager can resume the partition consumption. Note that when a record backs off the partition consumption gets paused for approximately that amount of time, so you must have a fixed backoff value per partition in order to make sure no record waits more than it should.

Since:

2.7

Author:

Tomaz Fernandes, Gary Russell

See Also:

SeekToCurrentErrorHandler

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static class	KafkaConsumerBackoffManager.Context	Provides the state that will be used for backing off.

Field Summary

Fields

Modifier and Type	Field	Description
static java.lang.String	INTERNAL_BACKOFF_CLOCK_BEAN_NAME	Internal Back Off Clock Bean Name.

Constructor Summary

Constructors

Constructor	Description
KafkaConsumerBackoffManager(ListenerContainerRegistry registry, java.time.Clock clock, org.springframework.core.task.TaskExecutor taskExecutor, org.springframework.retry.backoff.Sleeper sleeper)

Method Summary

Modifier and Type	Method	Description
protected void	addBackoff(KafkaConsumerBackoffManager.Context context, org.apache.kafka.common.TopicPartition topicPartition)
KafkaConsumerBackoffManager.Context	createContext(long dueTimestamp, java.lang.String listenerId, org.apache.kafka.common.TopicPartition topicPartition, org.apache.kafka.clients.consumer.Consumer<?,?> consumerForTimingCorrection)
protected KafkaConsumerBackoffManager.Context	getBackOffContext(org.apache.kafka.common.TopicPartition topicPartition)
void	maybeBackoff(KafkaConsumerBackoffManager.Context context)	Backs off if the current time is before the dueTimestamp provided in the KafkaConsumerBackoffManager.Context object.
void	onApplicationEvent(ListenerContainerPartitionIdleEvent partitionIdleEvent)
protected void	removeBackoff(org.apache.kafka.common.TopicPartition topicPartition)

Methods inherited from class java.lang.Object

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

Field Detail

INTERNAL_BACKOFF_CLOCK_BEAN_NAME

public static final java.lang.String INTERNAL_BACKOFF_CLOCK_BEAN_NAME

Internal Back Off Clock Bean Name.

See Also:

Constant Field Values

Constructor Detail

KafkaConsumerBackoffManager

public KafkaConsumerBackoffManager(ListenerContainerRegistry registry,
                                   @Qualifier("internalBackOffClock")
                                   java.time.Clock clock,
                                   org.springframework.core.task.TaskExecutor taskExecutor,
                                   org.springframework.retry.backoff.Sleeper sleeper)

Method Detail

maybeBackoff

public void maybeBackoff(KafkaConsumerBackoffManager.Context context)

Backs off if the current time is before the dueTimestamp provided in the KafkaConsumerBackoffManager.Context object.

Parameters:

context - the back off context for this execution.

onApplicationEvent

public void onApplicationEvent(ListenerContainerPartitionIdleEvent partitionIdleEvent)

Specified by:

onApplicationEvent in interface org.springframework.context.ApplicationListener<ListenerContainerPartitionIdleEvent>

addBackoff

protected void addBackoff(KafkaConsumerBackoffManager.Context context,
                          org.apache.kafka.common.TopicPartition topicPartition)

getBackOffContext

@Nullable
protected KafkaConsumerBackoffManager.Context getBackOffContext(org.apache.kafka.common.TopicPartition topicPartition)

removeBackoff

protected void removeBackoff(org.apache.kafka.common.TopicPartition topicPartition)

createContext

public KafkaConsumerBackoffManager.Context createContext(long dueTimestamp,
                                                         java.lang.String listenerId,
                                                         org.apache.kafka.common.TopicPartition topicPartition,
                                                         @Nullable
                                                         org.apache.kafka.clients.consumer.Consumer<?,?> consumerForTimingCorrection)