org.springframework.kafka.listener

Class DefaultAfterRollbackProcessor<K,V>

java.lang.Object

org.springframework.kafka.listener.FailedRecordProcessor

org.springframework.kafka.listener.DefaultAfterRollbackProcessor<K,V>

Type Parameters:

K - the key type.

V - the value type.

All Implemented Interfaces:

AfterRollbackProcessor<K,V>

public class DefaultAfterRollbackProcessor<K,V>
extends FailedRecordProcessor
implements AfterRollbackProcessor<K,V>

Default implementation of AfterRollbackProcessor. Seeks all topic/partitions so the records will be re-fetched, including the failed record. Starting with version 2.2 after a configurable number of failures for the same topic/partition/offset, that record will be skipped after calling a BiConsumer recoverer. The default recoverer simply logs the failed record.

Since:

1.3.5

Author:

Gary Russell

Field Summary

Fields inherited from class org.springframework.kafka.listener.FailedRecordProcessor

logger

Constructor Summary

Constructors

Constructor and Description
DefaultAfterRollbackProcessor() Construct an instance with the default recoverer which simply logs the record after (maxFailures) have occurred for a topic/partition/offset.
DefaultAfterRollbackProcessor(org.springframework.util.backoff.BackOff backOff) Construct an instance with the default recoverer which simply logs the record after the backOff returns STOP for a topic/partition/offset.
DefaultAfterRollbackProcessor(java.util.function.BiConsumer<org.apache.kafka.clients.consumer.ConsumerRecord<?,?>,java.lang.Exception> recoverer) Construct an instance with the provided recoverer which will be called after (maxFailures) have occurred for a topic/partition/offset.
DefaultAfterRollbackProcessor(java.util.function.BiConsumer<org.apache.kafka.clients.consumer.ConsumerRecord<?,?>,java.lang.Exception> recoverer, org.springframework.util.backoff.BackOff backOff) Construct an instance with the provided recoverer which will be called after the backOff returns STOP for a topic/partition/offset.
DefaultAfterRollbackProcessor(java.util.function.BiConsumer<org.apache.kafka.clients.consumer.ConsumerRecord<?,?>,java.lang.Exception> recoverer, int maxFailures) Deprecated. in favor of DefaultAfterRollbackProcessor(BackOff). IMPORTANT When using a FixedBackOff, the maxAttempts property represents retries (one less than maxFailures). To retry indefinitely, use a fixed or exponential BackOff configured appropriately. To use the other constructor with the semantics of this one, with maxFailures equal to 3, use new DefaultAfterRollbackProcessor(recoverer, new FixedBackOff(0L, 2L).
DefaultAfterRollbackProcessor(int maxFailures) Deprecated. in favor of DefaultAfterRollbackProcessor(BackOff). IMPORTANT When using a FixedBackOff, the maxAttempts property represents retries (one less than maxFailures). To retry indefinitely, use a fixed or exponential BackOff configured appropriately. To use the other constructor with the semantics of this one, with maxFailures equal to 3, use new DefaultAfterRollbackProcessor(new FixedBackOff(0L, 2L).

Method Summary

Modifier and Type	Method and Description
boolean	isProcessInTransaction() Return true to invoke AfterRollbackProcessor.process(List, Consumer, Exception, boolean) in a new transaction.
void	process(java.util.List<org.apache.kafka.clients.consumer.ConsumerRecord<K,V>> records, org.apache.kafka.clients.consumer.Consumer<K,V> consumer, java.lang.Exception exception, boolean recoverable) Process the remaining records.
void	setCommitRecovered(boolean commitRecovered) Set to true to commit the offset for a recovered record.
void	setKafkaTemplate(KafkaTemplate<K,V> kafkaTemplate) Set a KafkaTemplate to use to send the offset of a recovered record to a transaction.

Methods inherited from class org.springframework.kafka.listener.FailedRecordProcessor

addNotRetryableException, clearThreadState, getClassifier, getSkipPredicate, isCommitRecovered, removeNotRetryableException, setClassifications, setClassifier

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

clearThreadState

Constructor Detail

DefaultAfterRollbackProcessor

public DefaultAfterRollbackProcessor()

Construct an instance with the default recoverer which simply logs the record after (maxFailures) have occurred for a topic/partition/offset.

Since:

2.2

DefaultAfterRollbackProcessor

@Deprecated
public DefaultAfterRollbackProcessor(int maxFailures)

Deprecated. in favor of DefaultAfterRollbackProcessor(BackOff). IMPORTANT When using a FixedBackOff, the maxAttempts property represents retries (one less than maxFailures). To retry indefinitely, use a fixed or exponential BackOff configured appropriately. To use the other constructor with the semantics of this one, with maxFailures equal to 3, use new DefaultAfterRollbackProcessor(new FixedBackOff(0L, 2L).

Construct an instance with the default recoverer which simply logs the record after 'maxFailures' have occurred for a topic/partition/offset.

Parameters:

maxFailures - the maxFailures; a negative value is treated as infinity.

Since:

2.2.1

DefaultAfterRollbackProcessor

public DefaultAfterRollbackProcessor(org.springframework.util.backoff.BackOff backOff)

Construct an instance with the default recoverer which simply logs the record after the backOff returns STOP for a topic/partition/offset.

Parameters:

backOff - the BackOff.

Since:

2.3

DefaultAfterRollbackProcessor

public DefaultAfterRollbackProcessor(java.util.function.BiConsumer<org.apache.kafka.clients.consumer.ConsumerRecord<?,?>,java.lang.Exception> recoverer)

Construct an instance with the provided recoverer which will be called after (maxFailures) have occurred for a topic/partition/offset.

Parameters:

recoverer - the recoverer.

Since:

2.2

DefaultAfterRollbackProcessor

@Deprecated
public DefaultAfterRollbackProcessor(@Nullable
                                                 java.util.function.BiConsumer<org.apache.kafka.clients.consumer.ConsumerRecord<?,?>,java.lang.Exception> recoverer,
                                                 int maxFailures)

Deprecated. in favor of DefaultAfterRollbackProcessor(BackOff). IMPORTANT When using a FixedBackOff, the maxAttempts property represents retries (one less than maxFailures). To retry indefinitely, use a fixed or exponential BackOff configured appropriately. To use the other constructor with the semantics of this one, with maxFailures equal to 3, use new DefaultAfterRollbackProcessor(recoverer, new FixedBackOff(0L, 2L).

Construct an instance with the provided recoverer which will be called after maxFailures have occurred for a topic/partition/offset.

Parameters:

recoverer - the recoverer; if null, the default (logging) recoverer is used.

maxFailures - the maxFailures; a negative value is treated as infinity.

Since:

2.2

DefaultAfterRollbackProcessor

public DefaultAfterRollbackProcessor(@Nullable
                                     java.util.function.BiConsumer<org.apache.kafka.clients.consumer.ConsumerRecord<?,?>,java.lang.Exception> recoverer,
                                     org.springframework.util.backoff.BackOff backOff)

Construct an instance with the provided recoverer which will be called after the backOff returns STOP for a topic/partition/offset.

Parameters:

recoverer - the recoverer; if null, the default (logging) recoverer is used.

backOff - the BackOff.

Since:

2.3

Method Detail

process

public void process(java.util.List<org.apache.kafka.clients.consumer.ConsumerRecord<K,V>> records,
                    org.apache.kafka.clients.consumer.Consumer<K,V> consumer,
                    java.lang.Exception exception,
                    boolean recoverable)

Description copied from interface: AfterRollbackProcessor

Process the remaining records. Recoverable will be true if the container is processing individual records; this allows the processor to recover (skip) the failed record rather than re-seeking it. This is not possible with a batch listener since only the listener itself knows which record in the batch keeps failing. IMPORTANT: If invoked in a transaction when the listener was invoked with a single record, the transaction id will be based on the container group.id and the topic/partition of the failed record, to avoid issues with zombie fencing. So, generally, only its offset should be sent to the transaction. For other behavior the process method should manage its own transaction.

Specified by:

process in interface AfterRollbackProcessor<K,V>

Parameters:

records - the records.

consumer - the consumer.

exception - the exception

recoverable - the recoverable.

See Also:

AfterRollbackProcessor.isProcessInTransaction()

isProcessInTransaction

public boolean isProcessInTransaction()

Description copied from interface: AfterRollbackProcessor

Return true to invoke AfterRollbackProcessor.process(List, Consumer, Exception, boolean) in a new transaction. Because the container cannot infer the desired behavior, the processor is responsible for sending the offset to the transaction if it decides to skip the failing record.

Specified by:

isProcessInTransaction in interface AfterRollbackProcessor<K,V>

Returns:

true to run in a transaction; default false.

See Also:

AfterRollbackProcessor.process(List, Consumer, Exception, boolean)

setCommitRecovered

public void setCommitRecovered(boolean commitRecovered)

Set to true to commit the offset for a recovered record. Set to true and the container will run the process(List, Consumer, Exception, boolean) method in a transaction and, if a record is skipped and recovered, we will send its offset to the transaction. Requires a KafkaTemplate.

Overrides:

setCommitRecovered in class FailedRecordProcessor

Parameters:

commitRecovered - true to process in a transaction.

Since:

2.3

See Also:

isProcessInTransaction(), process(List, Consumer, Exception, boolean), setKafkaTemplate(KafkaTemplate)

setKafkaTemplate

public void setKafkaTemplate(KafkaTemplate<K,V> kafkaTemplate)

Set a KafkaTemplate to use to send the offset of a recovered record to a transaction.

Parameters:

kafkaTemplate - the template

Since:

2.2.5

See Also:

setCommitRecovered(boolean)