Class AbstractRecoveringExceptionHandler<R>

java.lang.Object

org.springframework.kafka.streams.AbstractRecoveringExceptionHandler<R>

Type Parameters:

R - the handler-specific response type

Direct Known Subclasses:

RecoveringDeserializationExceptionHandler, RecoveringProcessingExceptionHandler, RecoveringProductionExceptionHandler

public abstract class AbstractRecoveringExceptionHandler<R>
extends Object

Abstract base class for recovering Kafka Streams exception handlers.

Since:

4.1

Author:

Loïc Greffier

Field Summary

Fields

Modifier and Type	Field	Description
protected final DeadLetterRecordManager	deadLetterRecordManager
protected @Nullable String	deadLetterTopic
protected @Nullable KafkaStreamsDeadLetterDestinationResolver	destinationResolver
protected @Nullable ConsumerRecordRecoverer	recoverer

Constructor Summary

Constructors

Modifier	Constructor	Description
protected	AbstractRecoveringExceptionHandler()
protected	AbstractRecoveringExceptionHandler(ConsumerRecordRecoverer recoverer)

Method Summary

Modifier and Type	Method	Description
protected void	configureCommon(Map<String,?> configs, String destinationResolverKey, String recovererKey, @Nullable String legacyRecovererKey)	Configure common attributes.
protected abstract R	fail()	Create a response indicating that processing should fail.
protected R	handleErrorCommon(org.apache.kafka.streams.errors.ErrorHandlerContext context, org.apache.kafka.clients.consumer.ConsumerRecord<?,?> record, Exception exception)	Handles errors using a prioritized recovery strategy.
protected abstract R	resume(List<org.apache.kafka.clients.producer.ProducerRecord<byte[],byte[]>> deadLetterRecords)	Create a response indicating that processing should resume.

Methods inherited from class java.lang.Object

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

Field Details

deadLetterRecordManager

protected final DeadLetterRecordManager deadLetterRecordManager

recoverer

protected @Nullable ConsumerRecordRecoverer recoverer

destinationResolver

protected @Nullable KafkaStreamsDeadLetterDestinationResolver destinationResolver

deadLetterTopic

protected @Nullable String deadLetterTopic

Constructor Details

AbstractRecoveringExceptionHandler

protected AbstractRecoveringExceptionHandler()

AbstractRecoveringExceptionHandler

protected AbstractRecoveringExceptionHandler(ConsumerRecordRecoverer recoverer)

Method Details

handleErrorCommon

protected R handleErrorCommon(org.apache.kafka.streams.errors.ErrorHandlerContext context,
 org.apache.kafka.clients.consumer.ConsumerRecord<?,?> record,
 Exception exception)

Handles errors using a prioritized recovery strategy.

1. Resume the stream by forwarding to the topic-partition resolved by KafkaStreamsDeadLetterDestinationResolver via the native Kafka Streams DLQ.
2. Resume the stream by forwarding to the errors.dead.letter.queue.topic.name topic via the native Kafka Streams DLQ.
3. Delegate the recovery logic and dead-letter record sending to the ConsumerRecordRecoverer. Resume with no dead-letter records, as it is expected to be handled by the ConsumerRecordRecoverer.
4. Fail the stream with no dead-letter records.

Parameters:

context - the error handler context

record - the consumer record that caused the error

exception - the exception that occurred

Returns:

a handler-specific response

configureCommon

protected void configureCommon(Map<String,?> configs,
 String destinationResolverKey,
 String recovererKey,
 @Nullable String legacyRecovererKey)

Configure common attributes.

Parameters:

configs - the configuration map

destinationResolverKey - the property key for the native DLQ destination resolver

recovererKey - the property key for the recoverer

legacyRecovererKey - the legacy property key for the recoverer, used as fallback if recovererKey is not defined

fail

protected abstract R fail()

Create a response indicating that processing should fail.

Returns:

a handler-specific fail response

resume

protected abstract R resume(List<org.apache.kafka.clients.producer.ProducerRecord<byte[],byte[]>> deadLetterRecords)

Create a response indicating that processing should resume.

Parameters:

deadLetterRecords - The list of dead letter records to forward to DLQ

Returns:

a handler-specific resume response