Package org.springframework.kafka.listener

Class PartitionPausingBackoffManager

java.lang.Object

org.springframework.kafka.listener.PartitionPausingBackoffManager

All Implemented Interfaces:

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

public class PartitionPausingBackoffManager
extends java.lang.Object
implements KafkaConsumerBackoffManager, 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.

Since:

2.7

Author:

Tomaz Fernandes, Gary Russell

See Also:

SeekToCurrentErrorHandler

Nested Class Summary

Nested classes/interfaces inherited from interface org.springframework.kafka.listener.KafkaConsumerBackoffManager

KafkaConsumerBackoffManager.Context

Constructor Summary

Constructors

Constructor	Description
PartitionPausingBackoffManager(ListenerContainerRegistry listenerContainerRegistry)	Constructs an instance with the provided ListenerContainerRegistry and with no timing adjustment capabilities.
PartitionPausingBackoffManager(ListenerContainerRegistry listenerContainerRegistry, java.time.Clock clock)	Creates an instance with the provided ListenerContainerRegistry and Clock, with no timing adjustment capabilities.
PartitionPausingBackoffManager(ListenerContainerRegistry listenerContainerRegistry, KafkaConsumerTimingAdjuster kafkaConsumerTimingAdjuster)	Constructs an instance with the provided ListenerContainerRegistry and KafkaConsumerTimingAdjuster.
PartitionPausingBackoffManager(ListenerContainerRegistry listenerContainerRegistry, KafkaConsumerTimingAdjuster kafkaConsumerTimingAdjuster, java.time.Clock clock)	Creates an instance with the provided ListenerContainerRegistry, KafkaConsumerTimingAdjuster and Clock.

Method Summary

Modifier and Type	Method	Description
protected void	addBackoff(KafkaConsumerBackoffManager.Context context, org.apache.kafka.common.TopicPartition topicPartition)
void	backOffIfNecessary(KafkaConsumerBackoffManager.Context context)	Backs off if the current time is before the dueTimestamp provided in the KafkaConsumerBackoffManager.Context object.
protected KafkaConsumerBackoffManager.Context	getBackOffContext(org.apache.kafka.common.TopicPartition topicPartition)
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

Methods inherited from interface org.springframework.kafka.listener.KafkaConsumerBackoffManager

createContext

Constructor Detail

PartitionPausingBackoffManager

public PartitionPausingBackoffManager(ListenerContainerRegistry listenerContainerRegistry,
                                      KafkaConsumerTimingAdjuster kafkaConsumerTimingAdjuster)

Constructs an instance with the provided ListenerContainerRegistry and KafkaConsumerTimingAdjuster. The ListenerContainerRegistry is used to fetch the MessageListenerContainer that will be backed off / resumed. The KafkaConsumerTimingAdjuster is used to make timing adjustments in the message consumption so that it processes the message closer to its due time rather than later.

Parameters:

listenerContainerRegistry - the listenerContainerRegistry to use.

kafkaConsumerTimingAdjuster - the kafkaConsumerTimingAdjuster to use.

PartitionPausingBackoffManager

public PartitionPausingBackoffManager(ListenerContainerRegistry listenerContainerRegistry)

Constructs an instance with the provided ListenerContainerRegistry and with no timing adjustment capabilities. The ListenerContainerRegistry is used to fetch the MessageListenerContainer that will be backed off / resumed.

Parameters:

listenerContainerRegistry - the listenerContainerRegistry to use.

PartitionPausingBackoffManager

public PartitionPausingBackoffManager(ListenerContainerRegistry listenerContainerRegistry,
                                      KafkaConsumerTimingAdjuster kafkaConsumerTimingAdjuster,
                                      java.time.Clock clock)

Creates an instance with the provided ListenerContainerRegistry, KafkaConsumerTimingAdjuster and Clock.

Parameters:

listenerContainerRegistry - the listenerContainerRegistry to use.

kafkaConsumerTimingAdjuster - the kafkaConsumerTimingAdjuster to use.

clock - the clock to use.

PartitionPausingBackoffManager

public PartitionPausingBackoffManager(ListenerContainerRegistry listenerContainerRegistry,
                                      java.time.Clock clock)

Creates an instance with the provided ListenerContainerRegistry and Clock, with no timing adjustment capabilities.

Parameters:

listenerContainerRegistry - the listenerContainerRegistry to use.

clock - the clock to use.

Method Detail

backOffIfNecessary

public void backOffIfNecessary(KafkaConsumerBackoffManager.Context context)

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

Specified by:

backOffIfNecessary in interface KafkaConsumerBackoffManager

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)