处理异常 (Handling Exceptions) _ Spring for Apache Kafka3.2.2-SNAPSHOT中文文档(官方文档中文翻译)|Spring 教程 —

此版本仍在开发中，尚未被视为稳定版本。对于最新的稳定版本，请使用 Spring for Apache Kafka 3.2.1！

本节介绍如何处理在将 Spring 用于 Apache Kafka 时可能出现的各种异常。

侦听器错误处理程序

从版本 2.0 开始，批注具有一个新属性：.@KafkaListenererrorHandler

您可以使用来提供实现的 Bean 名称。此功能接口具有一种方法，如以下列表所示：errorHandlerKafkaListenerErrorHandler

@FunctionalInterface
public interface KafkaListenerErrorHandler {

    Object handleError(Message<?> message, ListenerExecutionFailedException exception) throws Exception;

}

您可以访问消息转换器生成的 spring-messaging 对象以及侦听器引发的异常，该异常包装在 . 错误处理程序可以抛出原始异常或新异常，这些异常被抛入容器。错误处理程序返回的任何内容都将被忽略。Message<?>ListenerExecutionFailedException

从版本 2.7 开始，您可以在和上设置属性，这会导致将原始内容添加到标头中的转换中。例如，如果您希望在侦听器错误处理程序中使用，这很有用。它可能用于请求/答复方案，在请求/答复方案中，您希望在捕获死信主题中的失败记录后，经过一定次数的重试后向发件人发送失败结果。rawRecordHeaderMessagingMessageConverterBatchMessagingMessageConverterConsumerRecordMessage<?>KafkaHeaders.RAW_DATADeadLetterPublishingRecoverer

@Bean
public KafkaListenerErrorHandler eh(DeadLetterPublishingRecoverer recoverer) {
    return (msg, ex) -> {
        if (msg.getHeaders().get(KafkaHeaders.DELIVERY_ATTEMPT, Integer.class) > 9) {
            recoverer.accept(msg.getHeaders().get(KafkaHeaders.RAW_DATA, ConsumerRecord.class), ex);
            return "FAILED";
        }
        throw ex;
    };
}

它有一个子接口（），可以通过以下方法访问使用者对象：ConsumerAwareListenerErrorHandler

Object handleError(Message<?> message, ListenerExecutionFailedException exception, Consumer<?, ?> consumer);

另一个子接口（）在使用手动 s 时提供对对象的访问。ManualAckListenerErrorHandlerAcknowledgmentAckMode

Object handleError(Message<?> message, ListenerExecutionFailedException exception,
			Consumer<?, ?> consumer, @Nullable Acknowledgment ack);

无论哪种情况，都不应对使用者执行任何搜索，因为容器不会知道它们。

容器错误处理程序

从 2.8 版开始，旧版和接口已被新的 . 这些错误处理程序可以处理记录侦听器和批处理侦听器的错误，从而允许单个侦听器容器工厂为这两种类型的侦听器创建容器。提供了替换大多数旧框架错误处理程序实现的实现。ErrorHandlerBatchErrorHandlerCommonErrorHandlerCommonErrorHandler

有关将自定义错误处理程序迁移到的信息，请参阅将自定义旧错误处理程序实现迁移到 CommonErrorHandler。CommonErrorHandler

使用事务时，默认情况下不会配置任何错误处理程序，因此异常将回滚事务。事务容器的错误处理由 AfterRollbackProcessor 处理。如果在使用事务时提供自定义错误处理程序，则在希望回滚事务时，它必须引发异常。

此接口具有一个默认方法，容器会调用该方法，以确定在错误处理程序返回而未引发异常的情况下是否应提交偏移量;默认情况下，它返回 true。isAckAfterHandle()

通常，当错误未被“处理”时（例如，在执行搜索操作之后），框架提供的错误处理程序将引发异常。默认情况下，此类异常由容器在级别记录。所有框架错误处理程序都进行了扩展，允许您控制记录这些异常的级别。ERRORKafkaExceptionLogLevelAware

/**
 * Set the level at which the exception thrown by this handler is logged.
 * @param logLevel the level (default ERROR).
 */
public void setLogLevel(KafkaException.Level logLevel) {
    ...
}

您可以指定要用于容器工厂中所有侦听器的全局错误处理程序。以下示例演示如何执行此操作：

@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
        kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
    ...
    factory.setCommonErrorHandler(myErrorHandler);
    ...
    return factory;
}

默认情况下，如果带注释的侦听器方法抛出异常，则将其抛向容器，并根据容器配置处理消息。

容器在调用错误处理程序之前提交任何挂起的偏移量提交。

如果您使用的是 Spring Boot，您只需将错误处理程序添加为 Boot，Boot 就会将其添加到自动配置的工厂中。@Bean

回退处理程序

错误处理程序（如 DefaultErrorHandler）使用 a 来确定在重试传递之前要等待多长时间。从版本 2.9 开始，您可以配置自定义 . 默认处理程序只是挂起线程，直到回退时间过去（或容器停止）。该框架还提供暂停侦听器容器，直到回退时间过去，然后恢复容器。当延迟时间长于消费者财产时，这很有用。请注意，实际回退时间的解析将受容器属性的影响。BackOffBackOffHandlerContainerPausingBackOffHandlermax.poll.interval.mspollTimeout

DefaultErrorHandler

这个新的错误处理程序取代了和，后者现在是多个版本的默认错误处理程序。一个区别是，批处理侦听器的回退行为（当引发除 a 以外的异常时）等同于重试完成批处理。SeekToCurrentErrorHandlerRecoveringBatchErrorHandlerBatchListenerFailedException

从版本 2.9 开始，可以配置为提供与查找未处理的记录偏移量相同的语义，如下所述，但实际上不查找。相反，记录由侦听器容器保留，并在错误处理程序退出后重新提交到侦听器（在执行单个暂停后，以保持使用者处于活动状态;如果使用非阻塞重试或 a，则暂停可能会延伸到多个轮询）。错误处理程序将结果返回到容器，该结果指示是否可以重新提交当前失败的记录，或者是否已恢复，然后不会再次将其发送到侦听器。若要启用此模式，请将属性设置为。DefaultErrorHandlerpoll()ContainerPausingBackOffHandlerseekAfterErrorfalse

错误处理程序可以恢复（跳过）不断失败的记录。默认情况下，在 10 次失败后，将记录失败的记录（在级别上）。您可以使用自定义恢复器（）和 a 来配置处理程序，以控制每个恢复器之间的传递尝试和延迟。使用 with 原因（有效地）无限重试。以下示例在三次尝试后配置恢复：ERRORBiConsumerBackOffFixedBackOffFixedBackOff.UNLIMITED_ATTEMPTS

DefaultErrorHandler errorHandler =
    new DefaultErrorHandler((record, exception) -> {
        // recover after 3 failures, with no back off - e.g. send to a dead-letter topic
    }, new FixedBackOff(0L, 2L));

若要使用此处理程序的自定义实例配置侦听器容器，请将其添加到容器工厂。

例如，使用容器工厂，可以添加如下内容：@KafkaListenerDefaultErrorHandler

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.getContainerProperties().setAckMode(AckMode.RECORD);
    factory.setCommonErrorHandler(new DefaultErrorHandler(new FixedBackOff(1000L, 2L)));
    return factory;
}

对于记录侦听器，这将重试最多 2 次（3 次传送尝试），并回退 1 秒，而不是默认配置（）。在重试用尽后，只会记录失败。FixedBackOff(0L, 9)

例如，如果返回 6 条记录（每个分区 0、1、2 中有 2 条），并且侦听器在第四个记录上引发异常，则容器通过提交前三条消息的偏移量来确认它们。寻求分区 1 的偏移量 1 和分区 2 的偏移量 0。下一个返回三个未处理的记录。pollDefaultErrorHandlerpoll()

如果为，则容器在调用错误处理程序之前提交前两个分区的偏移量。AckModeBATCH

对于批处理侦听器，侦听器必须抛出一个指示批处理中哪些记录失败的指示。BatchListenerFailedException

事件的顺序是：

在索引之前提交记录的偏移量。
如果重试未用尽，请执行查找，以便重新传递所有剩余记录（包括失败的记录）。
如果重试次数用尽，请尝试恢复失败的记录（仅限默认日志）并执行寻道，以便重新传递剩余的记录（不包括失败的记录）。已提交恢复的记录的偏移量。
如果重试已用尽且恢复失败，则将执行寻道，就好像重试未用尽一样。

从版本 2.9 开始，可以配置为提供与查找上述未处理的记录偏移量相同的语义，但实际上不查找。相反，错误处理程序创建一个仅包含未处理记录的新记录，然后将这些记录提交给侦听器（在执行单个暂停后，以保持使用者处于活动状态）。若要启用此模式，请将属性设置为。DefaultErrorHandlerConsumerRecords<?, ?>poll()seekAfterErrorfalse

默认恢复程序在重试耗尽后记录失败的记录。可以使用自定义恢复器，也可以使用框架提供的恢复器，例如 DeadLetterPublishingRecoverer。

使用 POJO 批处理侦听器（例如），并且您没有要添加到异常的完整使用者记录时，只需添加失败记录的索引：List<Thing>

@KafkaListener(id = "recovering", topics = "someTopic")
public void listen(List<Thing> things) {
    for (int i = 0; i < things.size(); i++) {
        try {
            process(things.get(i));
        }
        catch (Exception e) {
            throw new BatchListenerFailedException("Failed to process", i);
        }
    }
}

当容器配置为时，可以将错误处理程序配置为提交已恢复记录的偏移量;将属性设置为。AckMode.MANUAL_IMMEDIATEcommitRecoveredtrue

另请参阅发布死信记录。

使用事务时，. 请参阅回滚后处理器。DefaultAfterRollbackProcessor

认为某些异常是致命的，并且会跳过此类异常的重试;在第一次失败时调用恢复程序。默认情况下，被视为致命的例外情况包括：DefaultErrorHandler

DeserializationException
MessageConversionException
ConversionException
MethodArgumentResolutionException
NoSuchMethodException
ClassCastException

因为这些异常不太可能在重试传递时得到解决。

您可以向不可重试的类别添加更多异常类型，也可以完全替换分类异常的映射。有关更多信息，请参阅 Javadocs，以及 .DefaultErrorHandler.addNotRetryableException()DefaultErrorHandler.setClassifications()spring-retryBinaryExceptionClassifier

下面是一个添加到不可重试异常的示例：IllegalArgumentException

@Bean
public DefaultErrorHandler errorHandler(ConsumerRecordRecoverer recoverer) {
    DefaultErrorHandler handler = new DefaultErrorHandler(recoverer);
    handler.addNotRetryableExceptions(IllegalArgumentException.class);
    return handler;
}

错误处理程序可以配置一个或多个 s，接收重试和恢复进度的通知。从版本 2.8.10 开始，添加了批处理侦听器的方法。RetryListener

@FunctionalInterface
public interface RetryListener {

    void failedDelivery(ConsumerRecord<?, ?> record, Exception ex, int deliveryAttempt);

    default void recovered(ConsumerRecord<?, ?> record, Exception ex) {
    }

    default void recoveryFailed(ConsumerRecord<?, ?> record, Exception original, Exception failure) {
    }

    default void failedDelivery(ConsumerRecords<?, ?> records, Exception ex, int deliveryAttempt) {
    }

    default void recovered(ConsumerRecords<?, ?> records, Exception ex) {
    }

	default void recoveryFailed(ConsumerRecords<?, ?> records, Exception original, Exception failure) {
	}

}

有关更多信息，请参阅 JavaDocs。

如果恢复程序失败（引发异常），则失败的记录将包含在搜索中。如果恢复器失败，则默认情况下将重置，并且在再次尝试恢复之前，重新交付将再次进行回退。若要在恢复失败后跳过重试，请将错误处理程序设置为。BackOffresetStateOnRecoveryFailurefalse

您可以为错误处理程序提供一个，以根据失败的记录和/或异常确定要使用的内容：BiFunction<ConsumerRecord<?, ?>, Exception, BackOff>BackOff

handler.setBackOffFunction((record, ex) -> { ... });

如果函数返回，则将使用处理程序的默认值。nullBackOff

如果异常类型在两次失败之间发生更改，则设置为重试序列（包括选择新的，如果已配置）。当（版本 2.9 之前的默认值）时，不考虑异常类型。resetStateOnExceptionChangetrueBackOfffalse

从版本 2.9 开始，现在是默认设置。true

另请参阅传递尝试标头。

批处理错误处理程序的转换错误

从版本 2.8 开始，批处理侦听器现在可以正确处理转换错误，当将 a 与、 a 或 a 一起使用时。发生转换错误时，有效负载将设置为 null，并将反序列化异常添加到记录标头，类似于 . 侦听器中提供了 s 列表，因此侦听器可以抛出指示发生转换异常的第一个索引。MessageConverterByteArrayDeserializerBytesDeserializerStringDeserializerDefaultErrorHandlerErrorHandlingDeserializerConversionExceptionBatchListenerFailedException

例：

@KafkaListener(id = "test", topics = "topic")
void listen(List<Thing> in, @Header(KafkaHeaders.CONVERSION_FAILURES) List<ConversionException> exceptions) {
    for (int i = 0; i < in.size(); i++) {
        Foo foo = in.get(i);
        if (foo == null && exceptions.get(i) != null) {
            throw new BatchListenerFailedException("Conversion error", exceptions.get(i), i);
        }
        process(foo);
    }
}

重试完整批处理

现在，这是批处理侦听器的回退行为，其中侦听器抛出的异常不是 .DefaultErrorHandlerBatchListenerFailedException

无法保证在重新交付批次时，该批次具有相同数量的记录和/或重新交付的记录顺序相同。因此，不可能轻松维护批处理的重试状态。采用以下方法。如果批处理侦听器引发的异常不是，则从内存中的记录批处理中执行重试。为了避免在延长的重试序列期间重新平衡，错误处理程序会暂停使用者，在休眠之前轮询它以进行回退，每次重试，然后再次调用侦听器。如果/当重试次数用尽时，则为批处理中的每条记录调用。如果恢复程序引发异常，或者线程在休眠期间中断，则该批记录将在下一次轮询时重新传递。在退出之前，无论结果如何，消费者都会恢复。FallbackBatchErrorHandlerBatchListenerFailedExceptionConsumerRecordRecoverer

此机制不能用于事务。

在等待时间间隔时，错误处理程序将循环进行短暂的休眠，直到达到所需的延迟，同时检查容器是否已停止，允许休眠在停止后很快退出，而不是导致延迟。BackOffstop()

此机制不能用于事务。

容器停止错误处理程序

如果侦听器引发异常，则停止容器。对于记录侦听器，当为时，将提交已处理记录的偏移量。对于记录侦听器，当是任何手动值时，将提交已确认记录的偏移量。对于记录侦听器，当为时，或者对于批处理侦听器，在重新启动容器时重播整个批处理。CommonContainerStoppingErrorHandlerAckModeRECORDAckModeAckModeBATCH

容器停止后，将抛出包装的异常。这是为了使事务回滚（如果启用了事务）。ListenerExecutionFailedException

委派错误处理程序

可以委托给不同的错误处理程序，具体取决于异常类型。例如，您可能希望为大多数异常调用 a，或为其他异常调用 a。CommonDelegatingErrorHandlerDefaultErrorHandlerCommonContainerStoppingErrorHandler

所有委托必须共享相同的兼容属性（， ...）。ackAfterHandleseekAfterError

日志记录错误处理程序

只是记录异常;使用记录侦听器时，上一次轮询的剩余记录将传递给侦听器。对于批处理侦听器，将记录批处理中的所有记录。CommonLoggingErrorHandler

对记录侦听器和批处理侦听器使用不同的常见错误处理程序

如果您希望对记录侦听器和批处理侦听器使用不同的错误处理策略，则提供了允许为每种侦听器类型配置特定错误处理程序。CommonMixedErrorHandler

常见错误处理程序摘要

DefaultErrorHandler
CommonContainerStoppingErrorHandler
CommonDelegatingErrorHandler
CommonLoggingErrorHandler
CommonMixedErrorHandler

旧版错误处理程序及其替代程序

旧版错误处理程序更换

旧版错误处理程序	更换
`LoggingErrorHandler`	`CommonLoggingErrorHandler`
`BatchLoggingErrorHandler`	`CommonLoggingErrorHandler`
`ConditionalDelegatingErrorHandler`	`DelegatingErrorHandler`
`ConditionalDelegatingBatchErrorHandler`	`DelegatingErrorHandler`
`ContainerStoppingErrorHandler`	`CommonContainerStoppingErrorHandler`
`ContainerStoppingBatchErrorHandler`	`CommonContainerStoppingErrorHandler`
`SeekToCurrentErrorHandler`	`DefaultErrorHandler`
`SeekToCurrentBatchErrorHandler`	没有替代品，使用无限.`DefaultErrorHandlerBackOff`
`RecoveringBatchErrorHandler`	`DefaultErrorHandler`
`RetryingBatchErrorHandler`	没有替换，使用并抛出除以外的异常。`DefaultErrorHandlerBatchListenerFailedException`

LoggingErrorHandler

CommonLoggingErrorHandler

BatchLoggingErrorHandler

CommonLoggingErrorHandler

ConditionalDelegatingErrorHandler

DelegatingErrorHandler

ConditionalDelegatingBatchErrorHandler

DelegatingErrorHandler

ContainerStoppingErrorHandler

CommonContainerStoppingErrorHandler

ContainerStoppingBatchErrorHandler

CommonContainerStoppingErrorHandler

SeekToCurrentErrorHandler

DefaultErrorHandler

SeekToCurrentBatchErrorHandler

没有替代品，使用无限.DefaultErrorHandlerBackOff

RecoveringBatchErrorHandler

DefaultErrorHandler

RetryingBatchErrorHandler

没有替换，使用并抛出除以外的异常。DefaultErrorHandlerBatchListenerFailedException

将自定义旧版错误处理程序实现迁移到`CommonErrorHandler`

请参阅中的 JavaDocs。CommonErrorHandler

要替换 or 实现，应实现并保留返回（默认值）。您还应该实现以处理在记录处理范围之外发生的异常（例如使用者错误）。ErrorHandlerConsumerAwareErrorHandlerhandleOne()seeksAfterHandle()falsehandleOtherException()

若要替换实现，应实现并重写以返回（错误处理程序必须执行必要的查找）。您还应该实现 - 以处理在记录处理范围之外发生的异常（例如使用者错误）。RemainingRecordsErrorHandlerhandleRemaining()seeksAfterHandle()truehandleOtherException()

要替换任何实现，您还应该实现 - 以处理在记录处理范围之外发生的异常（例如使用者错误）。BatchErrorHandlerhandleBatch()handleOtherException()

旧版错误处理程序更换

旧版错误处理程序	更换
`LoggingErrorHandler`	`CommonLoggingErrorHandler`
`BatchLoggingErrorHandler`	`CommonLoggingErrorHandler`
`ConditionalDelegatingErrorHandler`	`DelegatingErrorHandler`
`ConditionalDelegatingBatchErrorHandler`	`DelegatingErrorHandler`
`ContainerStoppingErrorHandler`	`CommonContainerStoppingErrorHandler`
`ContainerStoppingBatchErrorHandler`	`CommonContainerStoppingErrorHandler`
`SeekToCurrentErrorHandler`	`DefaultErrorHandler`
`SeekToCurrentBatchErrorHandler`	没有替代品，使用无限.`DefaultErrorHandlerBackOff`
`RecoveringBatchErrorHandler`	`DefaultErrorHandler`
`RetryingBatchErrorHandler`	没有替换，使用并抛出除以外的异常。`DefaultErrorHandlerBatchListenerFailedException`

LoggingErrorHandler

CommonLoggingErrorHandler

BatchLoggingErrorHandler

CommonLoggingErrorHandler

ConditionalDelegatingErrorHandler

DelegatingErrorHandler

ConditionalDelegatingBatchErrorHandler

DelegatingErrorHandler

ContainerStoppingErrorHandler

CommonContainerStoppingErrorHandler

ContainerStoppingBatchErrorHandler

CommonContainerStoppingErrorHandler

SeekToCurrentErrorHandler

DefaultErrorHandler

SeekToCurrentBatchErrorHandler

没有替代品，使用无限.DefaultErrorHandlerBackOff

RecoveringBatchErrorHandler

DefaultErrorHandler

RetryingBatchErrorHandler

没有替换，使用并抛出除以外的异常。DefaultErrorHandlerBatchListenerFailedException

回滚后处理器

使用事务时，如果侦听器引发异常（并且错误处理程序（如果存在）引发异常），则事务将回滚。默认情况下，任何未处理的记录（包括失败的记录）都会在下次轮询时重新提取。这是通过在 . 使用批处理侦听器，将重新处理整个批次的记录（容器不知道批处理中的哪条记录失败）。若要修改此行为，可以使用自定义 . 例如，对于基于记录的侦听器，您可能希望跟踪失败的记录，并在尝试一定次数后放弃，可能将其发布到死信主题。seekDefaultAfterRollbackProcessorAfterRollbackProcessor

从版本 2.2 开始，现在可以恢复（跳过）不断失败的记录。默认情况下，在 10 次失败后，将记录失败的记录（在级别上）。您可以使用自定义恢复程序（）和最大故障数配置处理器。将属性设置为负数会导致无限重试。以下示例在三次尝试后配置恢复：DefaultAfterRollbackProcessorERRORBiConsumermaxFailures

AfterRollbackProcessor<String, String> processor =
    new DefaultAfterRollbackProcessor((record, exception) -> {
        // recover after 3 failures, with no back off - e.g. send to a dead-letter topic
    }, new FixedBackOff(0L, 2L));

当您不使用事务时，您可以通过配置 . 请参阅容器错误处理程序。DefaultErrorHandler

从版本 3.2 开始，Recovery 现在可以恢复（跳过）不断失败的整批记录。设置为启用此功能。ContainerProperties.setBatchRecoverAfterRollback(true)

默认行为，批处理侦听器无法恢复，因为框架不知道批处理中的哪条记录不断失败。在这种情况下，应用程序侦听器必须处理不断失败的记录。

另请参阅发布死信记录。

从版本 2.2.5 开始，可以在新事务中调用（在失败的事务回滚后启动）。然后，如果使用发布失败的记录，处理器会将恢复的记录在原始主题/分区中的偏移量发送到事务。若要启用此功能，请在 .DefaultAfterRollbackProcessorDeadLetterPublishingRecoverercommitRecoveredkafkaTemplateDefaultAfterRollbackProcessor

如果恢复程序失败（引发异常），则失败的记录将包含在搜索中。从版本 2.5.5 开始，如果恢复程序失败，则默认情况下将重置，并且在再次尝试恢复之前，将再次进行回退。在早期版本中，不会重置，并在下次失败时重新尝试恢复。若要恢复到以前的行为，请将处理器的属性设置为。BackOffBackOffresetStateOnRecoveryFailurefalse

从版本 2.6 开始，您现在可以为处理器提供一个，以根据失败的记录和/或异常来确定要使用的内容：BiFunction<ConsumerRecord<?, ?>, Exception, BackOff>BackOff

handler.setBackOffFunction((record, ex) -> { ... });

如果函数返回，则将使用处理器的默认值。nullBackOff

从版本 2.6.3 开始，如果异常类型在两次失败之间发生更改，则将重新启动重试序列（包括选择新的，如果已配置）。默认情况下，不考虑异常类型。resetStateOnExceptionChangetrueBackOff

从版本 2.3.1 开始，与类似，认为某些异常是致命的，并且对于此类异常会跳过重试;在第一次失败时调用恢复程序。默认情况下，被视为致命的例外情况包括：DefaultErrorHandlerDefaultAfterRollbackProcessor

DeserializationException
MessageConversionException
ConversionException
MethodArgumentResolutionException
NoSuchMethodException
ClassCastException

因为这些异常不太可能在重试传递时得到解决。

您可以向不可重试的类别添加更多异常类型，也可以完全替换分类异常的映射。有关更多信息，请参阅 Javadocs，以及 .DefaultAfterRollbackProcessor.setClassifications()spring-retryBinaryExceptionClassifier

下面是一个添加到不可重试异常的示例：IllegalArgumentException

@Bean
public DefaultAfterRollbackProcessor errorHandler(BiConsumer<ConsumerRecord<?, ?>, Exception> recoverer) {
    DefaultAfterRollbackProcessor processor = new DefaultAfterRollbackProcessor(recoverer);
    processor.addNotRetryableException(IllegalArgumentException.class);
    return processor;
}

另请参阅传递尝试标头。

使用 current 时，容器无法检测 a 是否由重新平衡引起，或者生产者是否因超时或到期而被撤销。因为在大多数情况下，它是由重新平衡引起的，所以容器不会调用（因为查找分区是不合适的，因为我们不再被分配它们）。如果您确保超时足够大以处理每个事务并定期执行“空”事务（例如通过），则可以避免由于超时和到期而导致的隔离。或者，可以将容器属性设置为并且容器将停止，从而避免丢失记录。可以使用 a 并检查属性来检测此情况。由于该事件还具有对容器的引用，因此可以使用此事件重新启动容器。kafka-clientsProducerFencedExceptiontransactional.idAfterRollbackProcessorListenerContainerIdleEventstopContainerWhenFencedtrueConsumerStoppedEventReasonFENCED

从版本 2.7 开始，在等待时间间隔时，错误处理程序将循环进行短暂的休眠，直到达到所需的延迟，同时检查容器是否已停止，允许休眠在停止后不久退出，而不是导致延迟。BackOffstop()

从版本 2.7 开始，处理器可以配置一个或多个 s，接收重试和恢复进度的通知。RetryListener

@FunctionalInterface
public interface RetryListener {

    void failedDelivery(ConsumerRecord<?, ?> record, Exception ex, int deliveryAttempt);

    default void recovered(ConsumerRecord<?, ?> record, Exception ex) {
    }

    default void recoveryFailed(ConsumerRecord<?, ?> record, Exception original, Exception failure) {
    }

}

有关更多信息，请参阅 JavaDocs。

默认行为，批处理侦听器无法恢复，因为框架不知道批处理中的哪条记录不断失败。在这种情况下，应用程序侦听器必须处理不断失败的记录。

传递尝试标头

以下内容仅适用于录制侦听器，不适用于批处理侦听器。

从版本 2.5 开始，当使用或实现时，可以启用将标头（）添加到记录中。此标头的值是从 1 开始递增的整数。当接收原始时，整数位于 .ErrorHandlerAfterRollbackProcessorDeliveryAttemptAwareKafkaHeaders.DELIVERY_ATTEMPTkafka_deliveryAttemptConsumerRecord<?, ?>byte[4]

int delivery = ByteBuffer.wrap(record.headers()
    .lastHeader(KafkaHeaders.DELIVERY_ATTEMPT).value())
    .getInt();

当与或一起使用时，可以通过将它作为参数添加到侦听器方法中来获得。@KafkaListenerDefaultKafkaHeaderMapperSimpleKafkaHeaderMapper@Header(KafkaHeaders.DELIVERY_ATTEMPT) int delivery

若要启用此标头的填充，请将容器属性设置为。默认情况下，它处于禁用状态，以避免查找每条记录的状态并添加标头的（小）开销。deliveryAttemptHeadertrue

和支持此功能。DefaultErrorHandlerDefaultAfterRollbackProcessor

侦听器信息标头

在某些情况下，能够知道侦听器在哪个容器中运行是很有用的。

从版本 2.8.4 开始，您现在可以在侦听器容器上设置属性，或在注释上设置属性。然后，容器会在标头中将其添加到所有传入消息中;然后，它可以用于记录拦截器、过滤器等，或用于侦听器本身。listenerInfoinfo@KafkaListenerKafkaListener.LISTENER_INFO

@KafkaListener(id = "something", topics = "topic", filter = "someFilter",
        info = "this is the something listener")
public void listen(@Payload Thing thing,
        @Header(KafkaHeaders.LISTENER_INFO) String listenerInfo) {
    ...
}

在 or 实现中使用时，标头作为字节数组在使用者记录中，使用的属性进行转换。RecordInterceptorRecordFilterStrategyKafkaListenerAnnotationBeanPostProcessorcharSet

标头映射器也会在从使用者记录创建时转换为，并且从不将此标头映射到出站记录上。StringMessageHeaders

对于 POJO 批处理侦听器，从版本 2.8.6 开始，标头将复制到批处理的每个成员中，并且在转换后也可以作为单个参数使用。String

@KafkaListener(id = "list2", topics = "someTopic", containerFactory = "batchFactory",
        info = "info for batch")
public void listen(List<Thing> list,
        @Header(KafkaHeaders.RECEIVED_KEY) List<Integer> keys,
        @Header(KafkaHeaders.RECEIVED_PARTITION) List<Integer> partitions,
        @Header(KafkaHeaders.RECEIVED_TOPIC) List<String> topics,
        @Header(KafkaHeaders.OFFSET) List<Long> offsets,
        @Header(KafkaHeaders.LISTENER_INFO) String info) {
            ...
}

如果批处理侦听器具有筛选器，并且筛选器生成空批处理，则需要添加到参数中，因为该信息不适用于空批处理。required = false@Header

如果您收到的信息在每个 .List<Message<Thing>>KafkaHeaders.LISTENER_INFOMessage<?>

有关使用批处理的更多信息，请参阅批处理侦听器。

如果批处理侦听器具有筛选器，并且筛选器生成空批处理，则需要添加到参数中，因为该信息不适用于空批处理。required = false@Header

发布死信记录

当记录达到最大故障数时，可以使用记录恢复器配置 and。该框架提供了，它将失败的消息发布到另一个主题。恢复程序需要一个，用于发送记录。您也可以选择使用，调用来解析目标主题和分区。DefaultErrorHandlerDefaultAfterRollbackProcessorDeadLetterPublishingRecovererKafkaTemplate<Object, Object>BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition>

默认情况下，死信记录将发送到名为（原始主题名称后缀为）的主题，并发送到与原始记录相同的分区。因此，使用默认解析程序时，死信主题的分区数必须至少与原始主题相同。<originalTopic>.DLT.DLT

如果返回的分区为负，则未在中设置该分区，因此 Kafka 会选择该分区。从版本 2.2.4 开始，任何（例如，在方法中检测到异常时引发）都会使用该属性进行增强。这允许目标解析程序使用它，以及 to select the dead letter 主题中的信息。TopicPartitionProducerRecordListenerExecutionFailedException@KafkaListenergroupIdConsumerRecord

以下示例演示如何连接自定义目标解析程序：

DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template,
        (r, e) -> {
            if (e instanceof FooException) {
                return new TopicPartition(r.topic() + ".Foo.failures", r.partition());
            }
            else {
                return new TopicPartition(r.topic() + ".other.failures", r.partition());
            }
        });
CommonErrorHandler errorHandler = new DefaultErrorHandler(recoverer, new FixedBackOff(0L, 2L));

发送到死信主题的记录通过以下标头得到增强：

KafkaHeaders.DLT_EXCEPTION_FQCN：Exception 类名（通常为，但也可以是其他）。ListenerExecutionFailedException
KafkaHeaders.DLT_EXCEPTION_CAUSE_FQCN：异常原因类名（如果存在）（从版本 2.8 开始）。
KafkaHeaders.DLT_EXCEPTION_STACKTRACE：异常堆栈跟踪。
KafkaHeaders.DLT_EXCEPTION_MESSAGE：异常消息。
KafkaHeaders.DLT_KEY_EXCEPTION_FQCN：Exception 类名（仅限密钥反序列化错误）。
KafkaHeaders.DLT_KEY_EXCEPTION_STACKTRACE：异常堆栈跟踪（仅限关键反序列化错误）。
KafkaHeaders.DLT_KEY_EXCEPTION_MESSAGE：异常消息（仅限关键反序列化错误）。
KafkaHeaders.DLT_ORIGINAL_TOPIC：原始主题。
KafkaHeaders.DLT_ORIGINAL_PARTITION：原始分区。
KafkaHeaders.DLT_ORIGINAL_OFFSET：原始偏移量。
KafkaHeaders.DLT_ORIGINAL_TIMESTAMP：原始时间戳。
KafkaHeaders.DLT_ORIGINAL_TIMESTAMP_TYPE：原始时间戳类型。
KafkaHeaders.DLT_ORIGINAL_CONSUMER_GROUP：无法处理记录的原始使用者组（从版本 2.8 开始）。

关键异常仅由 s 引起，因此没有 .DeserializationExceptionDLT_KEY_EXCEPTION_CAUSE_FQCN

有两种机制可以添加更多标头。

子类化恢复器并覆盖 - 调用并添加更多标头。createProducerRecord()super.createProducerRecord()
提供 a 接收消费者记录和异常，返回一个对象;那里的标头将被复制到最终的生产者记录中;另请参阅管理死信记录标头。用于设置 .BiFunctionHeaderssetHeadersFunction()BiFunction

第二个更易于实现，但第一个具有更多信息，包括已组装的标准标头。

从版本 2.3 开始，当与结合使用时，发布者会将死信生产者记录中的记录还原为无法反序列化的原始值。以前，为 null，用户代码必须从消息标头中解码。此外，您可以向发布者提供多个 s;例如，如果要发布 from a 以及使用不同序列化程序从成功反序列化的记录中的值，则可能需要这样做。下面是使用使用 and 序列化程序的 s 配置发布者的示例：ErrorHandlingDeserializervalue()value()DeserializationExceptionKafkaTemplatebyte[]DeserializationExceptionKafkaTemplateStringbyte[]

@Bean
public DeadLetterPublishingRecoverer publisher(KafkaTemplate<?, ?> stringTemplate,
        KafkaTemplate<?, ?> bytesTemplate) {
    Map<Class<?>, KafkaTemplate<?, ?>> templates = new LinkedHashMap<>();
    templates.put(String.class, stringTemplate);
    templates.put(byte[].class, bytesTemplate);
    return new DeadLetterPublishingRecoverer(templates);
}

发布者使用地图键查找适合即将发布的模板。建议使用 A，以便按顺序检查密钥。value()LinkedHashMap

当发布值并且有多个模板时，恢复器将为类查找模板;如果不存在，则将使用中的第一个模板。nullVoidvalues().iterator()

从 2.7 开始，您可以使用该方法，以便在消息发布失败时引发异常。您还可以设置超时，以便使用验证发件人是否成功。setFailIfSendResultIsErrorsetWaitForSendResultTimeout

如果恢复程序失败（引发异常），则失败的记录将包含在搜索中。从版本 2.5.5 开始，如果恢复程序失败，则默认情况下将重置，并且在再次尝试恢复之前，将再次进行回退。在早期版本中，不会重置，并在下次失败时重新尝试恢复。若要恢复到以前的行为，请将错误处理程序的属性设置为。BackOffBackOffresetStateOnRecoveryFailurefalse

从版本 2.3 开始，恢复器还可以与 Kafka Streams 一起使用 - 有关详细信息，请参阅从反序列化异常中恢复。

在标头和（使用 Java 序列化）中添加反序列化异常。默认情况下，这些标头不会保留在发布到死信主题的消息中。从版本 2.7 开始，如果键和值都未通过反序列化，则两者的原始值都将填充在发送到 DLT 的记录中。ErrorHandlingDeserializerErrorHandlingDeserializer.VALUE_DESERIALIZER_EXCEPTION_HEADERErrorHandlingDeserializer.KEY_DESERIALIZER_EXCEPTION_HEADER

如果传入的记录相互依赖，但可能无序到达，则将失败的记录重新发布到原始主题的尾部（多次）可能会很有用，而不是将其直接发送到死信主题。有关示例，请参阅此堆栈溢出问题。

以下错误处理程序配置将完全执行此操作：

@Bean
public ErrorHandler eh(KafkaOperations<String, String> template) {
    return new DefaultErrorHandler(new DeadLetterPublishingRecoverer(template,
            (rec, ex) -> {
                org.apache.kafka.common.header.Header retries = rec.headers().lastHeader("retries");
                if (retries == null) {
                    retries = new RecordHeader("retries", new byte[] { 1 });
                    rec.headers().add(retries);
                }
                else {
                    retries.value()[0]++;
                }
                return retries.value()[0] > 5
                        ? new TopicPartition("topic.DLT", rec.partition())
                        : new TopicPartition("topic", rec.partition());
            }), new FixedBackOff(0L, 0L));
}

从版本 2.7 开始，恢复程序将检查目标解析程序选择的分区是否确实存在。如果分区不存在，则中的分区设置为，允许选择分区。可以通过将属性设置为来禁用此检查。ProducerRecordnullKafkaProducerverifyPartitionfalse

从版本 3.1 开始，将属性设置为将记录恢复记录和异常。logRecoveryRecordtrue

管理死信记录标头

请参阅上面的发布死信记录，该属性有两个属性，用于在标头已存在时管理标头（例如，在重新处理失败的死信记录时，包括使用非阻塞重试时）。DeadLetterPublishingRecoverer

appendOriginalHeaders（默认true)
stripPreviousExceptionHeaders（自 2.8 版起默认）true

Apache Kafka 支持多个同名的标头;要获取“最新”值，可以使用;若要获取多个标头上的迭代器，请使用 .headers.lastHeader(headerName)headers.headers(headerName).iterator()

重复重新发布失败的记录时，这些标头可能会增长（并最终导致发布失败，因为）;对于异常标头，尤其是堆栈跟踪标头，尤其如此。RecordTooLargeException

这两个属性的原因是，虽然您可能只想保留最后一个异常信息，但您可能希望保留记录在每次失败时传递的主题的历史记录。

appendOriginalHeaders应用于所有命名的标头，而应用于所有名为的标头。ORIGINALstripPreviousExceptionHeadersEXCEPTION

从版本 2.8.4 开始，您现在可以控制将哪些标准标头添加到输出记录中。有关默认添加的（当前）10 个标准标头的通用名称，请参阅（这些不是实际的标头名称，只是一个抽象;实际的标头名称由子类可以覆盖的方法设置。enum HeadersToAddgetHeaderNames()

若要排除标头，请使用以下方法;例如，若要禁止在标头中添加异常堆栈跟踪，请使用：excludeHeaders()

DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template);
recoverer.excludeHeaders(HeaderNames.HeadersToAdd.EX_STACKTRACE);

此外，您可以通过添加 ;这还会禁用所有标准异常标头。ExceptionHeadersCreator

DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template);
recoverer.setExceptionHeadersCreator((kafkaHeaders, exception, isKey, headerNames) -> {
    kafkaHeaders.add(new RecordHeader(..., ...));
});

同样从版本 2.8.4 开始，您现在可以通过该方法提供多个标头函数。这允许应用其他函数，即使已经注册了另一个函数，例如，在使用非阻塞重试时也是如此。addHeadersFunction

另请参阅使用非阻塞重试进行故障标头管理。

`ExponentialBackOffWithMaxRetries`实现

Spring Framework 提供了许多实现。默认情况下，将无限期重试;要在重试一定次数后放弃，需要计算 . 从 2.7.3 版本开始，Spring for Apache Kafka 提供了 which 是一个子类，它接收属性并自动计算，这稍微方便一些。BackOffExponentialBackOffmaxElapsedTimeExponentialBackOffWithMaxRetriesmaxRetriesmaxElapsedTime

@Bean
DefaultErrorHandler handler() {
    ExponentialBackOffWithMaxRetries bo = new ExponentialBackOffWithMaxRetries(6);
    bo.setInitialInterval(1_000L);
    bo.setMultiplier(2.0);
    bo.setMaxInterval(10_000L);
    return new DefaultErrorHandler(myRecoverer, bo);
}

这将在几秒钟后重试，然后再调用恢复程序。1, 2, 4, 8, 10, 10

侦听器错误处理程序

容器错误处理程序

回退处理程序

DefaultErrorHandler

批处理错误处理程序的转换错误

重试完整批处理

容器停止错误处理程序

委派错误处理程序

日志记录错误处理程序

对记录侦听器和批处理侦听器使用不同的常见错误处理程序

常见错误处理程序摘要

旧版错误处理程序及其替代程序

将自定义旧版错误处理程序实现迁移到CommonErrorHandler

回滚后处理器

传递尝试标头

侦听器信息标头

发布死信记录

管理死信记录标头

ExponentialBackOffWithMaxRetries实现

将自定义旧版错误处理程序实现迁移到`CommonErrorHandler`

`ExponentialBackOffWithMaxRetries`实现