Package org.springframework.kafka.listener.adapter

Class RetryingMessageListenerAdapter<K,V>

java.lang.Object

org.springframework.kafka.listener.adapter.AbstractDelegatingMessageListenerAdapter<T>

org.springframework.kafka.listener.adapter.AbstractRetryingMessageListenerAdapter<K,V,MessageListener<K,V>>

org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter<K,V>

Type Parameters:

K - the key type.

V - the value type.

All Implemented Interfaces:

AcknowledgingConsumerAwareMessageListener<K,V>, ConsumerSeekAware, DelegatingMessageListener<MessageListener<K,V>>, GenericMessageListener<org.apache.kafka.clients.consumer.ConsumerRecord<K,V>>, MessageListener<K,V>

public class RetryingMessageListenerAdapter<K,V>
extends AbstractRetryingMessageListenerAdapter<K,V,MessageListener<K,V>>
implements AcknowledgingConsumerAwareMessageListener<K,V>

A retrying message listener adapter for MessageListeners.

Author:

Gary Russell

Nested Class Summary

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

ConsumerSeekAware.ConsumerSeekCallback

Field Summary

Fields

Modifier and Type	Field	Description
static java.lang.String	CONTEXT_ACKNOWLEDGMENT	RetryContext attribute key for an acknowledgment if the listener is capable of acknowledging.
static java.lang.String	CONTEXT_CONSUMER	RetryContext attribute key for the consumer if the listener is consumer-aware.
static java.lang.String	CONTEXT_RECORD	RetryContext attribute key for the record.

Fields inherited from class org.springframework.kafka.listener.adapter.AbstractDelegatingMessageListenerAdapter

delegate, delegateType, logger

Constructor Summary

Constructors

Constructor	Description
RetryingMessageListenerAdapter(MessageListener<K,V> messageListener, org.springframework.retry.support.RetryTemplate retryTemplate)	Construct an instance with the provided template and delegate.
RetryingMessageListenerAdapter(MessageListener<K,V> messageListener, org.springframework.retry.support.RetryTemplate retryTemplate, org.springframework.retry.RecoveryCallback<? extends java.lang.Object> recoveryCallback)	Construct an instance with the provided template, callback and delegate.
RetryingMessageListenerAdapter(MessageListener<K,V> messageListener, org.springframework.retry.support.RetryTemplate retryTemplate, org.springframework.retry.RecoveryCallback<? extends java.lang.Object> recoveryCallback, boolean stateful)	Construct an instance with the provided template, callback and delegate.

Method Summary

Modifier and Type	Method	Description
void	onMessage(org.apache.kafka.clients.consumer.ConsumerRecord<K,V> data)	Invoked with data from kafka.
void	onMessage(org.apache.kafka.clients.consumer.ConsumerRecord<K,V> data, org.apache.kafka.clients.consumer.Consumer<?,?> consumer)	Invoked with data from kafka and provides access to the Consumer.
void	onMessage(org.apache.kafka.clients.consumer.ConsumerRecord<K,V> data, Acknowledgment acknowledgment)	Invoked with data from kafka.
void	onMessage(org.apache.kafka.clients.consumer.ConsumerRecord<K,V> record, Acknowledgment acknowledgment, org.apache.kafka.clients.consumer.Consumer<?,?> consumer)	Invoked with data from kafka and provides access to the Consumer.

Methods inherited from class org.springframework.kafka.listener.adapter.AbstractRetryingMessageListenerAdapter

getRecoveryCallback, getRetryTemplate

Methods inherited from class org.springframework.kafka.listener.adapter.AbstractDelegatingMessageListenerAdapter

getDelegate, onIdleContainer, onPartitionsAssigned, onPartitionsRevoked, registerSeekCallback

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.ConsumerSeekAware

unregisterSeekCallback

Field Detail

CONTEXT_ACKNOWLEDGMENT

public static final java.lang.String CONTEXT_ACKNOWLEDGMENT

RetryContext attribute key for an acknowledgment if the listener is capable of acknowledging.

See Also:

Constant Field Values

CONTEXT_CONSUMER

public static final java.lang.String CONTEXT_CONSUMER

RetryContext attribute key for the consumer if the listener is consumer-aware.

See Also:

Constant Field Values

CONTEXT_RECORD

public static final java.lang.String CONTEXT_RECORD

RetryContext attribute key for the record.

See Also:

Constant Field Values

Constructor Detail

RetryingMessageListenerAdapter

public RetryingMessageListenerAdapter(MessageListener<K,V> messageListener,
                                      org.springframework.retry.support.RetryTemplate retryTemplate)

Construct an instance with the provided template and delegate. The exception will be thrown to the container after retries are exhausted.

Parameters:

messageListener - the delegate listener.

retryTemplate - the template.

RetryingMessageListenerAdapter

public RetryingMessageListenerAdapter(MessageListener<K,V> messageListener,
                                      org.springframework.retry.support.RetryTemplate retryTemplate,
                                      org.springframework.retry.RecoveryCallback<? extends java.lang.Object> recoveryCallback)

Construct an instance with the provided template, callback and delegate.

Parameters:

messageListener - the delegate listener.

retryTemplate - the template.

recoveryCallback - the recovery callback; if null, the exception will be thrown to the container after retries are exhausted.

RetryingMessageListenerAdapter

public RetryingMessageListenerAdapter(MessageListener<K,V> messageListener,
                                      org.springframework.retry.support.RetryTemplate retryTemplate,
                                      org.springframework.retry.RecoveryCallback<? extends java.lang.Object> recoveryCallback,
                                      boolean stateful)

Construct an instance with the provided template, callback and delegate. When using stateful retry, the retry context key is a concatenated String topic-partition-offset. A SeekToCurrentErrorHandler is required in the listener container because stateful retry will throw the exception to the container for each delivery attempt.

Parameters:

messageListener - the delegate listener.

retryTemplate - the template.

recoveryCallback - the recovery callback; if null, the exception will be thrown to the container after retries are exhausted.

stateful - true for stateful retry.

Since:

2.1.3

Method Detail

onMessage

public void onMessage(org.apache.kafka.clients.consumer.ConsumerRecord<K,V> record,
                      Acknowledgment acknowledgment,
                      org.apache.kafka.clients.consumer.Consumer<?,?> consumer)

Description copied from interface: GenericMessageListener

Invoked with data from kafka and provides access to the Consumer. The default implementation throws UnsupportedOperationException.

Specified by:

onMessage in interface AcknowledgingConsumerAwareMessageListener<K,V>

Specified by:

onMessage in interface GenericMessageListener<K>

Parameters:

record - the data to be processed.

acknowledgment - the acknowledgment.

consumer - the consumer.

onMessage

public void onMessage(org.apache.kafka.clients.consumer.ConsumerRecord<K,V> data)

Description copied from interface: AcknowledgingConsumerAwareMessageListener

Invoked with data from kafka. Containers should never call this since it they will detect that we are a consumer aware acknowledging listener.

Specified by:

onMessage in interface AcknowledgingConsumerAwareMessageListener<K,V>

Specified by:

onMessage in interface GenericMessageListener<K>

Parameters:

data - the data to be processed.

onMessage

public void onMessage(org.apache.kafka.clients.consumer.ConsumerRecord<K,V> data,
                      Acknowledgment acknowledgment)

Description copied from interface: GenericMessageListener

Invoked with data from kafka. The default implementation throws UnsupportedOperationException.

Specified by:

onMessage in interface GenericMessageListener<K>

Parameters:

data - the data to be processed.

acknowledgment - the acknowledgment.

onMessage

public void onMessage(org.apache.kafka.clients.consumer.ConsumerRecord<K,V> data,
                      org.apache.kafka.clients.consumer.Consumer<?,?> consumer)

Description copied from interface: GenericMessageListener

Invoked with data from kafka and provides access to the Consumer. The default implementation throws UnsupportedOperationException.

Specified by:

onMessage in interface GenericMessageListener<K>

Parameters:

data - the data to be processed.

consumer - the consumer.