Class DefaultAfterRollbackProcessor<K,V>
java.lang.Object
org.springframework.kafka.listener.KafkaExceptionLogLevelAware
org.springframework.kafka.listener.ExceptionClassifier
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> DeliveryAttemptAware
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, Francois Rosiere, Wang Zhiyang
-
Field Summary
Fields inherited from class org.springframework.kafka.listener.FailedRecordProcessor
logger
-
Constructor Summary
ConstructorsConstructorDescriptionConstruct an instance with the default recoverer which simply logs the record after 10 (maxFailures) have occurred for a topic/partition/offset.DefaultAfterRollbackProcessor
(BiConsumer<org.apache.kafka.clients.consumer.ConsumerRecord<?, ?>, Exception> recoverer) Construct an instance with the provided recoverer which will be called after 10 (maxFailures) have occurred for a topic/partition/offset.DefaultAfterRollbackProcessor
(BiConsumer<org.apache.kafka.clients.consumer.ConsumerRecord<?, ?>, Exception> recoverer, BackOff backOff) Construct an instance with the provided recoverer which will be called after the backOff returns STOP for a topic/partition/offset.DefaultAfterRollbackProcessor
(BiConsumer<org.apache.kafka.clients.consumer.ConsumerRecord<?, ?>, Exception> recoverer, BackOff backOff, KafkaOperations<?, ?> kafkaOperations, boolean commitRecovered) Construct an instance with the provided recoverer which will be called after the backOff returns STOP for a topic/partition/offset.DefaultAfterRollbackProcessor
(BiConsumer<org.apache.kafka.clients.consumer.ConsumerRecord<?, ?>, Exception> recoverer, BackOff backOff, BackOffHandler backOffHandler, KafkaOperations<?, ?> kafkaOperations, boolean commitRecovered) Construct an instance with the provided recoverer which will be called after the backOff returns STOP for a topic/partition/offset.DefaultAfterRollbackProcessor
(BackOff backOff) Construct an instance with the default recoverer which simply logs the record after the backOff returns STOP for a topic/partition/offset. -
Method Summary
Modifier and TypeMethodDescriptionvoid
Optional method to clear thread state; will be called just before a consumer thread terminates.boolean
Return true to invokeAfterRollbackProcessor.process(List, Consumer, MessageListenerContainer, Exception, boolean, ContainerProperties.EOSMode)
in a new transaction.void
process
(List<org.apache.kafka.clients.consumer.ConsumerRecord<K, V>> records, org.apache.kafka.clients.consumer.Consumer<K, V> consumer, MessageListenerContainer container, Exception exception, boolean recoverable, ContainerProperties.EOSMode eosMode) Process the remaining records.void
processBatch
(org.apache.kafka.clients.consumer.ConsumerRecords<K, V> records, List<org.apache.kafka.clients.consumer.ConsumerRecord<K, V>> recordList, org.apache.kafka.clients.consumer.Consumer<K, V> consumer, MessageListenerContainer container, Exception exception, boolean recoverable, ContainerProperties.EOSMode eosMode) Process the entire batch of records.Methods inherited from class org.springframework.kafka.listener.FailedRecordProcessor
deliveryAttempt, getFailureTracker, getRetryListeners, isCommitRecovered, isSeekAfterError, setBackOffFunction, setCommitRecovered, setResetStateOnExceptionChange, setResetStateOnRecoveryFailure, setRetryListeners, setSeekAfterError
Methods inherited from class org.springframework.kafka.listener.ExceptionClassifier
addNotRetryableExceptions, addRetryableExceptions, defaultFalse, defaultFalse, defaultFatalExceptionsList, getClassifier, notRetryable, removeClassification, setClassifications
Methods inherited from class org.springframework.kafka.listener.KafkaExceptionLogLevelAware
getLogLevel, setLogLevel
-
Constructor Details
-
DefaultAfterRollbackProcessor
public DefaultAfterRollbackProcessor()Construct an instance with the default recoverer which simply logs the record after 10 (maxFailures) have occurred for a topic/partition/offset.- Since:
- 2.2
-
DefaultAfterRollbackProcessor
Construct an instance with the default recoverer which simply logs the record after the backOff returns STOP for a topic/partition/offset.- Parameters:
backOff
- theBackOff
.- Since:
- 2.3
-
DefaultAfterRollbackProcessor
public DefaultAfterRollbackProcessor(BiConsumer<org.apache.kafka.clients.consumer.ConsumerRecord<?, ?>, Exception> recoverer) Construct an instance with the provided recoverer which will be called after 10 (maxFailures) have occurred for a topic/partition/offset.- Parameters:
recoverer
- the recoverer.- Since:
- 2.2
-
DefaultAfterRollbackProcessor
public DefaultAfterRollbackProcessor(@Nullable BiConsumer<org.apache.kafka.clients.consumer.ConsumerRecord<?, ?>, Exception> recoverer, 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
- theBackOff
.- Since:
- 2.3
-
DefaultAfterRollbackProcessor
public DefaultAfterRollbackProcessor(@Nullable BiConsumer<org.apache.kafka.clients.consumer.ConsumerRecord<?, ?>, Exception> recoverer, BackOff backOff, @Nullable KafkaOperations<?, ?> kafkaOperations, boolean commitRecovered) 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
- theBackOff
.kafkaOperations
- for sending the recovered offset to the transaction.commitRecovered
- true to commit the recovered record's offset; requires aKafkaOperations
.- Since:
- 2.5.3
-
DefaultAfterRollbackProcessor
public DefaultAfterRollbackProcessor(@Nullable BiConsumer<org.apache.kafka.clients.consumer.ConsumerRecord<?, ?>, Exception> recoverer, BackOff backOff, @Nullable BackOffHandler backOffHandler, @Nullable KafkaOperations<?, ?> kafkaOperations, boolean commitRecovered) 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
- theBackOff
.backOffHandler
- theBackOffHandler
.kafkaOperations
- for sending the recovered offset to the transaction.commitRecovered
- true to commit the recovered record's offset; requires aKafkaOperations
.- Since:
- 2.9
-
-
Method Details
-
process
public void process(List<org.apache.kafka.clients.consumer.ConsumerRecord<K, V>> records, org.apache.kafka.clients.consumer.Consumer<K, V> consumer, @Nullable MessageListenerContainer container, Exception exception, boolean recoverable, ContainerProperties.EOSMode eosMode) 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 interfaceAfterRollbackProcessor<K,
V> - Parameters:
records
- the records.consumer
- the consumer.container
- the container.exception
- the exceptionrecoverable
- the recoverable.eosMode
- theContainerProperties.EOSMode
.- See Also:
-
processBatch
public void processBatch(org.apache.kafka.clients.consumer.ConsumerRecords<K, V> records, List<org.apache.kafka.clients.consumer.ConsumerRecord<K, V>> recordList, org.apache.kafka.clients.consumer.Consumer<K, V> consumer, @Nullable MessageListenerContainer container, Exception exception, boolean recoverable, ContainerProperties.EOSMode eosMode) Description copied from interface:AfterRollbackProcessor
Process the entire batch of records. Recoverable will be true if the container is processing entire batch of records;- Specified by:
processBatch
in interfaceAfterRollbackProcessor<K,
V> - Parameters:
records
- the records.recordList
- the record list.consumer
- the consumer.container
- the container.exception
- the exceptionrecoverable
- the recoverable.eosMode
- theContainerProperties.EOSMode
.- See Also:
-
isProcessInTransaction
public boolean isProcessInTransaction()Description copied from interface:AfterRollbackProcessor
Return true to invokeAfterRollbackProcessor.process(List, Consumer, MessageListenerContainer, Exception, boolean, ContainerProperties.EOSMode)
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 interfaceAfterRollbackProcessor<K,
V> - Returns:
- true to run in a transaction; default false.
- See Also:
-
clearThreadState
public void clearThreadState()Description copied from interface:AfterRollbackProcessor
Optional method to clear thread state; will be called just before a consumer thread terminates.- Specified by:
clearThreadState
in interfaceAfterRollbackProcessor<K,
V> - Overrides:
clearThreadState
in classFailedRecordProcessor
-