Package org.springframework.kafka.listener

Class FailedBatchProcessor

java.lang.Object

org.springframework.kafka.listener.KafkaExceptionLogLevelAware

org.springframework.kafka.listener.ExceptionClassifier

org.springframework.kafka.listener.FailedRecordProcessor

org.springframework.kafka.listener.FailedBatchProcessor

All Implemented Interfaces:

DeliveryAttemptAware

Direct Known Subclasses:

DefaultErrorHandler

public abstract class FailedBatchProcessor
extends FailedRecordProcessor

Subclass of FailedRecordProcessor that can process (and recover) a batch. If the listener throws a BatchListenerFailedException, the offsets prior to the failed record are committed and the remaining records have seeks performed. When the retries are exhausted, the failed record is sent to the recoverer instead of being included in the seeks. If other exceptions are thrown, the fallback handler takes the processing.

Since:

2.8

Author:

Gary Russell, Francois Rosiere, Wang Zhiyang, Artem Bilan

Field Summary

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

logger

Constructor Summary

Constructors

Constructor	Description
FailedBatchProcessor(BiConsumer<org.apache.kafka.clients.consumer.ConsumerRecord<?,?>,Exception> recoverer, BackOff backOff, BackOffHandler backOffHandler, CommonErrorHandler fallbackHandler)	Construct an instance with the provided properties.
FailedBatchProcessor(BiConsumer<org.apache.kafka.clients.consumer.ConsumerRecord<?,?>,Exception> recoverer, BackOff backOff, CommonErrorHandler fallbackHandler)	Construct an instance with the provided properties.

Method Summary

Modifier and Type	Method	Description
protected void	doHandle(Exception thrownException, org.apache.kafka.clients.consumer.ConsumerRecords<?,?> data, org.apache.kafka.clients.consumer.Consumer<?,?> consumer, MessageListenerContainer container, Runnable invokeListener)
protected CommonErrorHandler	getFallbackBatchHandler()	Return the fallback batch error handler.
protected <K, V> org.apache.kafka.clients.consumer.ConsumerRecords<K,V>	handle(Exception thrownException, org.apache.kafka.clients.consumer.ConsumerRecords<?,?> data, org.apache.kafka.clients.consumer.Consumer<?,?> consumer, MessageListenerContainer container, Runnable invokeListener)
protected void	notRetryable(Stream<Class<? extends Exception>> notRetryable)	Subclasses can override this to receive notification of configuration of not retryable exceptions.
Boolean	removeClassification(Class<? extends Exception> exceptionType)	Remove an exception type from the configured list.
void	setClassifications(Map<Class<? extends Throwable>,Boolean> classifications, boolean defaultValue)	Set an exception classifications to determine whether the exception should cause a retry (until exhaustion) or not.
void	setLogLevel(KafkaException.Level logLevel)	Set the level at which the exception thrown by this handler is logged.
void	setReclassifyOnExceptionChange(boolean reclassifyOnExceptionChange)	Set to false to not reclassify the exception if different from the previous failure.
void	setRetryListeners(RetryListener... listeners)	Set one or more RetryListener to receive notifications of retries and recovery.

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

clearThreadState, deliveryAttempt, getFailureTracker, getRetryListeners, isCommitRecovered, isSeekAfterError, setBackOffFunction, setCommitRecovered, setResetStateOnExceptionChange, setResetStateOnRecoveryFailure, setSeekAfterError

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

addNotRetryableExceptions, addRetryableExceptions, defaultFalse, defaultFalse, defaultFatalExceptionsList, getClassifier

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

getLogLevel

Methods inherited from class java.lang.Object

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

Constructor Details

FailedBatchProcessor

public FailedBatchProcessor(@Nullable
 BiConsumer<org.apache.kafka.clients.consumer.ConsumerRecord<?,?>,Exception> recoverer,
 BackOff backOff,
 CommonErrorHandler fallbackHandler)

Construct an instance with the provided properties.

Parameters:

recoverer - the recoverer.

backOff - the back off.

fallbackHandler - the fallback handler.

FailedBatchProcessor

public FailedBatchProcessor(@Nullable
 BiConsumer<org.apache.kafka.clients.consumer.ConsumerRecord<?,?>,Exception> recoverer,
 BackOff backOff,
 @Nullable
 BackOffHandler backOffHandler,
 CommonErrorHandler fallbackHandler)

Construct an instance with the provided properties.

Parameters:

recoverer - the recoverer.

backOff - the back off.

backOffHandler - the BackOffHandler

fallbackHandler - the fallback handler.

Since:

2.9

Method Details

setRetryListeners

public void setRetryListeners(RetryListener... listeners)

Description copied from class: FailedRecordProcessor

Set one or more RetryListener to receive notifications of retries and recovery.

Overrides:

setRetryListeners in class FailedRecordProcessor

Parameters:

listeners - the listeners.

setLogLevel

public void setLogLevel(KafkaException.Level logLevel)

Description copied from class: KafkaExceptionLogLevelAware

Set the level at which the exception thrown by this handler is logged.

Overrides:

setLogLevel in class KafkaExceptionLogLevelAware

Parameters:

logLevel - the level (default ERROR).

setReclassifyOnExceptionChange

public void setReclassifyOnExceptionChange(boolean reclassifyOnExceptionChange)

Set to false to not reclassify the exception if different from the previous failure. If the changed exception is classified as retryable, the existing back off sequence is used; a new sequence is not started. Default true. Only applies when the fallback batch error handler (for exceptions other than BatchListenerFailedException) is the default.

Parameters:

reclassifyOnExceptionChange - false to not reclassify.

Since:

2.9.7

notRetryable

protected void notRetryable(Stream<Class<? extends Exception>> notRetryable)

Description copied from class: ExceptionClassifier

Subclasses can override this to receive notification of configuration of not retryable exceptions.

Overrides:

notRetryable in class ExceptionClassifier

Parameters:

notRetryable - the not retryable exceptions.

setClassifications

public void setClassifications(Map<Class<? extends Throwable>,Boolean> classifications,
 boolean defaultValue)

Description copied from class: ExceptionClassifier

Set an exception classifications to determine whether the exception should cause a retry (until exhaustion) or not. If not, we go straight to the recoverer. By default, the following exceptions will not be retried:

• DeserializationException
• MessageConversionException
• ConversionException
• MethodArgumentResolutionException
• NoSuchMethodException
• ClassCastException

All others will be retried. When calling this method, the defaults will not be applied.

Overrides:

setClassifications in class ExceptionClassifier

Parameters:

classifications - the classifications.

defaultValue - whether or not to retry non-matching exceptions.

See Also:

• BinaryExceptionClassifier(Map, boolean)
• ExceptionClassifier.addNotRetryableExceptions(Class...)

removeClassification

@Nullable
public Boolean removeClassification(Class<? extends Exception> exceptionType)

Description copied from class: ExceptionClassifier

Remove an exception type from the configured list. By default, the following exceptions will not be retried:

• DeserializationException
• MessageConversionException
• ConversionException
• MethodArgumentResolutionException
• NoSuchMethodException
• ClassCastException

All others will be retried, unless ExceptionClassifier.defaultFalse() has been called.

Overrides:

removeClassification in class ExceptionClassifier

Parameters:

exceptionType - the exception type.

Returns:

the classification of the exception if removal was successful; null otherwise.

See Also:

• ExceptionClassifier.addNotRetryableExceptions(Class...)
• ExceptionClassifier.setClassifications(Map, boolean)

getFallbackBatchHandler

protected CommonErrorHandler getFallbackBatchHandler()

Return the fallback batch error handler.

Returns:

the handler.

Since:

2.8.8

doHandle

protected void doHandle(Exception thrownException,
 org.apache.kafka.clients.consumer.ConsumerRecords<?,?> data,
 org.apache.kafka.clients.consumer.Consumer<?,?> consumer,
 MessageListenerContainer container,
 Runnable invokeListener)

handle

protected <K,
V>
org.apache.kafka.clients.consumer.ConsumerRecords<K,V> handle(Exception thrownException,
 org.apache.kafka.clients.consumer.ConsumerRecords<?,?> data,
 org.apache.kafka.clients.consumer.Consumer<?,?> consumer,
 MessageListenerContainer container,
 Runnable invokeListener)