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 (starting from the failed one) are re-sought for redelivery. When retries are exhausted, the failed record is sent to the recoverer instead of being included in the seeks. Any other exception type is delegated to the fallback handler.

Note: committing offsets for preceding records assumes those records were fully and irreversibly processed before the exception was thrown. Batch listeners that process records in parallel, use multi-step pipelines, or run inside a transaction should throw a plain RuntimeException instead of BatchListenerFailedException if they cannot guarantee that; doing so causes the full batch to be redelivered rather than silently skipping records whose side effects may not have completed.

Since:

2.8

See Also:

• BatchListenerFailedException

Field Summary

Fields inherited from class FailedRecordProcessor

logger

Constructor Summary

Constructors

Constructor	Description
FailedBatchProcessor(@Nullable BiConsumer<org.apache.kafka.clients.consumer.ConsumerRecord<?,?>, Exception> recoverer, org.springframework.util.backoff.BackOff backOff, @Nullable BackOffHandler backOffHandler, CommonErrorHandler fallbackHandler)	Construct an instance with the provided properties.
FailedBatchProcessor(@Nullable BiConsumer<org.apache.kafka.clients.consumer.ConsumerRecord<?,?>, Exception> recoverer, org.springframework.util.backoff.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.
@Nullable 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	setResetStateOnExceptionChange(boolean resetStateOnExceptionChange)	Set to true to reset the retry BackOff if the exception is a different type to the previous failure for the same record.
void	setRetryListeners(RetryListener... listeners)	Set one or more RetryListener to receive notifications of retries and recovery.

Methods inherited from class FailedRecordProcessor

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

Methods inherited from class ExceptionClassifier

addNotRetryableExceptions, addRetryableExceptions, defaultFalse, defaultFalse, defaultFatalExceptionsList, getExceptionMatcher

Methods inherited from class KafkaExceptionLogLevelAware

getLogLevel

Methods inherited from class 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,
 org.springframework.util.backoff.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,
 org.springframework.util.backoff.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. Default true. If the changed exception is classified as retryable, the existing back off sequence is used if resetStateOnExceptionChange is false; a new sequence is started if resetStateOnExceptionChange is 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

setResetStateOnExceptionChange

public void setResetStateOnExceptionChange(boolean resetStateOnExceptionChange)

Description copied from class: FailedRecordProcessor

Set to true to reset the retry BackOff if the exception is a different type to the previous failure for the same record. The backOffFunction, if provided, will be called to get the BackOff to use for the new exception; otherwise, the configured BackOff will be used. Default true since 2.9; set to false to use the existing retry state, even when exceptions change.

Overrides:

setResetStateOnExceptionChange in class FailedRecordProcessor

Parameters:

resetStateOnExceptionChange - true to reset.

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 to retry non-matching exceptions.

See Also:

• ExceptionClassifier.addNotRetryableExceptions(Class...)

removeClassification

public @Nullable 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)