版本：4.1.1

本站内所有文档均为中英对照文档，点击中文可以显示英文。此提醒连续关闭5次后，将不再显示。

文档内容来源于 spring.io，由 springdoc.tech 翻译，版权归属于 SPRING.IO (Broadcom. Inc)。可供个人学习、研究，未经许可，不得进行转载或用于商业行为。

处理异常

Doubao 中英对照 Handling Exceptions

本节描述在使用 Spring for Apache Kafka 时如何处理可能出现的各种异常。

监听器错误处理器

从 2.0 版本开始，@KafkaListener 注解新增了一个属性：errorHandler。

你可以使用 errorHandler 来提供 KafkaListenerErrorHandler 实现的 bean 名称。该函数式接口有一个方法，如下清单所示：

@FunctionalInterface
public interface KafkaListenerErrorHandler {

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

}

你可以访问由消息转换器产生的 spring-messaging Message<?> 对象，以及由监听器抛出、被包装在 ListenerExecutionFailedException 中的异常。错误处理器可以抛出原始异常或新的异常，该异常会被抛给容器。错误处理器返回的任何值都会被忽略。

从 2.7 版本开始，你可以在 MessagingMessageConverter 和 BatchMessagingMessageConverter 上设置 rawRecordHeader 属性，这会导致原始 ConsumerRecord 以 KafkaHeaders.RAW_DATA 头添加到转换后的 Message<?> 中。例如，如果你想在监听器错误处理器中使用 DeadLetterPublishingRecoverer，这会很有用。它可能用于请求/回复场景：在将失败记录捕获到死信主题之后、经过若干次重试后，你希望向发送方发送失败结果。

@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);

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

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

无论哪种情况，你都不应在消费者上执行任何 seek，因为容器不会察觉这些 seek。

容器错误处理器

从 2.8 版本开始，旧版 ErrorHandler 和 BatchErrorHandler 接口已被新的 CommonErrorHandler 取代。这些错误处理器可以同时处理记录监听器和批量监听器的错误，从而允许单个监听容器工厂为两种类型的监听器创建容器。提供了 CommonErrorHandler 实现来取代大多数旧版框架错误处理器实现。

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

使用事务时，默认不配置任何错误处理器，以便异常回滚事务。事务性容器的错误处理由回滚后处理器处理。如果在使用事务时提供了自定义错误处理器，而你希望事务回滚，则它必须抛出异常。

该接口有一个默认方法 isAckAfterHandle()，容器调用它来确定如果错误处理器没有抛出异常就返回，是否应提交偏移量；它默认返回 true。

通常，当错误未被 "处理"（例如执行 seek 操作之后）时，框架提供的错误处理器会抛出异常。默认情况下，此类异常由容器以 ERROR 级别记录。所有框架错误处理器都继承 KafkaExceptionLogLevelAware，它允许你控制这些异常被记录的级别。

/**
 * 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，只需将错误处理器作为 @Bean 添加，Boot 就会把它添加到自动配置的工厂中。

退避处理器

诸如 DefaultErrorHandler 之类的错误处理器使用 BackOff 来确定在重试投递之前等待多久。从 2.9 版本开始，你可以配置自定义的 BackOffHandler。默认处理器只是挂起线程，直到退避时间过去（或容器被停止）。该框架还提供了 ContainerPausingBackOffHandler，它会暂停监听容器直到退避时间过去，然后恢复容器。当延迟长于 max.poll.interval.ms 消费者属性时，这会很有用。请注意，实际退避时间的精度会受到 pollTimeout 容器属性的影响。

DefaultErrorHandler

这个新的错误处理器取代了 SeekToCurrentErrorHandler 和 RecoveringBatchErrorHandler，后两者在数个版本中一直是默认错误处理器。一个区别是，批量监听器的回退行为（当抛出的不是 BatchListenerFailedException 时）等同于重试整个批。

important

从 2.9 版本开始，DefaultErrorHandler 可以配置为提供与下文所述 seek 未处理记录偏移量相同的语义，但不实际执行 seek。相反，记录由监听容器保留，并在错误处理器退出后重新提交给监听器（并在执行一次暂停的 poll() 之后，以保持消费者存活；如果正在使用非阻塞重试或 ContainerPausingBackOffHandler，该暂停可能跨越多次 poll）。错误处理器向容器返回一个结果，指示当前失败记录是否可以重新提交，或者它是否已被恢复、从而不再发送给监听器。要启用此模式，将属性 seekAfterError 设置为 false。

错误处理器可以恢复（跳过）一条不断失败的记录。默认情况下，在十次失败后，失败记录会被记录（以 ERROR 级别）。你可以用自定义的恢复器（BiConsumer）和一个 BackOff 来配置该处理器，后者控制投递尝试次数以及每次之间的延迟。使用带有 FixedBackOff.UNLIMITED_ATTEMPTS 的 FixedBackOff 会（实际上）导致无限重试。以下示例配置了三次尝试后恢复：

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));

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

例如，对于 @KafkaListener 容器工厂，你可以如下添加 DefaultErrorHandler：

@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;
}

对于记录监听器，这将以 1 秒的退避重试投递最多 2 次（共 3 次投递尝试），而不是默认配置（FixedBackOff(0L, 9)）。重试用尽后失败仅被记录。

举例来说，如果 poll 返回 6 条记录（分区 0、1、2 各两条），而监听器在第四条记录上抛出异常，容器通过提交前三条消息的偏移量来确认它们。DefaultErrorHandler 对分区 1 seek 到偏移量 1，对分区 2 seek 到偏移量 0。下一次 poll() 返回三条未处理的记录。

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

对于批量监听器，监听器必须抛出一个 BatchListenerFailedException，指明批中哪些记录失败。

事件序列为：

提交索引之前记录的偏移量。
如果重试未用尽，执行 seek，以便所有剩余记录（包括失败记录）被重新投递。
如果重试已用尽，尝试恢复失败记录（默认仅记录），并执行 seek，以便剩余记录（不含失败记录）被重新投递。被恢复记录的偏移量被提交。
如果重试已用尽且恢复失败，则像重试未用尽那样执行 seek。

important

从 2.9 版本开始，DefaultErrorHandler 可以配置为提供与上文所述 seek 未处理记录偏移量相同的语义，但不实际执行 seek。相反，错误处理器会创建一个仅包含未处理记录的新 ConsumerRecords<?, ?>，然后将其提交给监听器（并在执行一次暂停的 poll() 之后，以保持消费者存活）。要启用此模式，将属性 seekAfterError 设置为 false。

默认恢复器在重试用尽后记录失败记录。你可以使用自定义恢复器，或框架提供的恢复器，例如 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_IMMEDIATE 时，错误处理器可以配置为提交被恢复记录的偏移量；将 commitRecovered 属性设置为 true。

另请参阅发布死信记录。

使用事务时，类似功能由 DefaultAfterRollbackProcessor 提供。请参阅回滚后处理器。

DefaultErrorHandler 将某些异常视为致命，对此类异常跳过重试；恢复器在第一次失败时被调用。默认被视为致命的异常有：

DeserializationException
MessageConversionException
ConversionException
MethodArgumentResolutionException
NoSuchMethodException
ClassCastException

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

你可以向不可重试类别添加更多异常类型，或者完全替换已分类异常的映射。更多信息请参阅 DefaultErrorHandler.addNotRetryableException() 和 DefaultErrorHandler.setClassifications() 的 Javadocs，以及 ExceptionMatcher。

以下示例向不可重试异常添加了 IllegalArgumentException：

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

important

DefaultErrorHandler 只处理继承自 RuntimeException 的异常。继承自 Error 的异常完全绕过错误处理器，导致消费者立即终止、关闭 Kafka 连接，并跳过所有重试/恢复机制。这一关键区别意味着，尽管消费者已终止、不再处理消息，应用程序仍可能报告健康状态。务必确保消息处理代码中抛出的异常显式继承自 RuntimeException 而非 Error，以便进行适当的错误处理。换句话说，如果应用程序抛出异常，请确保它继承自 RuntimeException，而不是不经意间继承自 Error。诸如 OutOfMemoryError、IllegalAccessError 以及应用程序无法控制的其他错误等标准错误仍被视为 Error，不会重试。

错误处理器可以配置一个或多个 RetryListener，接收重试和恢复进度的通知。从 2.8.10 版本开始，新增了用于批量监听器的方法。

@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。

important

如果恢复器失败（抛出异常），失败记录将被包含在 seek 中。如果恢复器失败，默认会重置 BackOff，在再次尝试恢复之前，重新投递将再次经历各次退避。要在恢复失败后跳过重试，将错误处理器的 resetStateOnRecoveryFailure 设置为 false。

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

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

如果该函数返回 null，将使用处理器的默认 BackOff。

将 resetStateOnExceptionChange 设置为 true，那么如果两次失败之间异常类型发生变化，重试序列将重新开始（包括在如此配置时选择新的 BackOff）。当为 false（2.9 版本之前的默认值）时，不考虑异常类型。

从 2.9 版本开始，此项默认为 true。

另请参阅投递尝试头。

使用死信主题的批量监听器错误处理

important

批量监听器不支持非阻塞重试（@RetryableTopic 注解）。对于批量监听器的死信主题功能，请将 DefaultErrorHandler 与 DeadLetterPublishingRecoverer 一起使用。

使用 BatchListenerFailedException

为指明批中具体哪条记录失败，抛出一个 BatchListenerFailedException：

@KafkaListener(id = "batch-listener", topics = "myTopic", containerFactory = "batchFactory")
public void listen(List<ConsumerRecord<String, Order>> records) {
    for (ConsumerRecord<String, Order> record : records) {
        try {
            process(record.value());
        }
        catch (Exception e) {
            // Identifies the failed record for error handling
            throw new BatchListenerFailedException("Failed to process", e, record);
        }
    }
}

对于没有 ConsumerRecord 的 POJO 批量监听器，请改用索引：

@KafkaListener(id = "batch-listener", topics = "myTopic", containerFactory = "batchFactory")
public void listen(List<Order> orders) {
    for (int i = 0; i < orders.size(); i++) {
        try {
            process(orders.get(i));
        }
        catch (Exception e) {
            throw new BatchListenerFailedException("Failed to process", e, i);
        }
    }
}

为批量监听器配置死信主题

在你的批量监听器容器工厂上，用 DeadLetterPublishingRecoverer 配置一个 DefaultErrorHandler：

@Bean
public ConcurrentKafkaListenerContainerFactory<String, Order> batchFactory(
        ConsumerFactory<String, Order> consumerFactory,
        KafkaTemplate<String, Order> kafkaTemplate) {

    ConcurrentKafkaListenerContainerFactory<String, Order> factory =
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory);
    factory.setBatchListener(true);

    // Configure Dead Letter Publishing
    DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(kafkaTemplate,
            (record, ex) -> new TopicPartition(record.topic() + "-dlt", record.partition()));

    // Configure retries: 3 attempts with 1 second between each
    DefaultErrorHandler errorHandler = new DefaultErrorHandler(recoverer,
            new FixedBackOff(1000L, 2L)); // 2 retries = 3 total attempts

    factory.setCommonErrorHandler(errorHandler);
    return factory;
}

批量错误处理的工作方式

当抛出 BatchListenerFailedException 时，DefaultErrorHandler：

提交失败记录之前所有记录的偏移量
根据 BackOff 配置重试失败记录（及后续记录）
重试用尽时发布到 DLT——只有失败记录被发送到 DLT
提交失败记录的偏移量，并重新投递剩余记录进行处理

以下是一个 6 条记录的批、索引 2 处的记录失败的示例流程：

第一次尝试：记录 0、1 处理成功；记录 2 失败
容器提交记录 0、1 的偏移量
重试第 1 次：重试记录 2、3、4、5
重试第 2 次：再次重试记录 2、3、4、5
重试用尽后：记录 2 被发布到 DLT，其偏移量被提交
容器继续处理记录 3、4、5

对特定异常跳过重试

默认情况下，DefaultErrorHandler 会重试除致命异常（如 DeserializationException、MessageConversionException 等）之外的所有异常。要为你自己的异常类型跳过重试，请用异常分类配置错误处理器。

错误处理器会检查 BatchListenerFailedException 的 cause，以确定是否应跳过重试：

@Bean
public ConcurrentKafkaListenerContainerFactory<String, Order> batchFactory(
        ConsumerFactory<String, Order> consumerFactory,
        KafkaTemplate<String, Order> kafkaTemplate) {

    ConcurrentKafkaListenerContainerFactory<String, Order> factory =
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory);
    factory.setBatchListener(true);

    DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(kafkaTemplate);
    DefaultErrorHandler errorHandler = new DefaultErrorHandler(recoverer,
            new FixedBackOff(1000L, 2L));

    // Add custom exception types that should skip retries and go directly to DLT
    errorHandler.addNotRetryableExceptions(ValidationException.class, InvalidFormatException.class);

    factory.setCommonErrorHandler(errorHandler);
    return factory;
}

现在在你的监听器中：

@KafkaListener(id = "batch-listener", topics = "orders", containerFactory = "batchFactory")
public void processOrders(List<ConsumerRecord<String, Order>> records) {
    for (ConsumerRecord<String, Order> record : records) {
        try {
            process(record.value());
        }
        catch (DatabaseException e) {
            // Will be retried 3 times (according to BackOff configuration)
            throw new BatchListenerFailedException("Database error", e, record);
        }
        catch (ValidationException e) {
            // Skips retries - goes directly to DLT
            // (because ValidationException is configured as not retryable)
            throw new BatchListenerFailedException("Validation failed", e, record);
        }
    }
}

important

错误处理器会检查 BatchListenerFailedException 的 cause（第二个参数）。如果 cause 被分类为不可重试，该记录会被立即发送到 DLT，不重试。

偏移量提交行为

理解偏移量提交对批量错误处理很重要：

AckMode.BATCH（批量监听器最常见）：
- 失败记录之前的偏移量在错误处理之前提交
- 失败记录的偏移量在成功恢复（DLT 发布）之后提交
AckMode.MANUAL_IMMEDIATE：
- 设置 errorHandler.setCommitRecovered(true) 以提交被恢复记录的偏移量
- 你在监听器中控制确认时机

手动确认的示例：

@KafkaListener(id = "manual-batch", topics = "myTopic", containerFactory = "manualBatchFactory")
public void listen(List<ConsumerRecord<String, Order>> records, Acknowledgment ack) {
    for (ConsumerRecord<String, Order> record : records) {
        try {
            process(record.value());
        }
        catch (Exception e) {
            throw new BatchListenerFailedException("Processing failed", e, record);
        }
    }
    ack.acknowledge();
}

批量错误处理器的转换错误

从 2.8 版本开始，当使用带有 ByteArrayDeserializer、BytesDeserializer 或 StringDeserializer 的 MessageConverter 以及 DefaultErrorHandler 时，批量监听器现在可以正确处理转换错误。当发生转换错误时，载荷被置为 null，并在记录头中添加一个反序列化异常，类似于 ErrorHandlingDeserializer。监听器中提供了一个 ConversionException 列表，以便监听器可以抛出一个 BatchListenerFailedException，指明发生转换异常的第一个索引。

示例：

@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);
    }
}

批量监听器的反序列化错误

important

批量监听器需要 手动处理 反序列化错误。与记录监听器不同，没有自动错误处理器来检测反序列化失败并将其路由到 DLT。你必须显式检查失败记录并抛出 BatchListenerFailedException。

使用 ErrorHandlingDeserializer 以防止反序列化失败停止整个批：

@Bean
public ConsumerFactory<String, Order> consumerFactory() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

    // Wrap your deserializer with ErrorHandlingDeserializer
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
    props.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class.getName());

    return new DefaultKafkaConsumerFactory<>(props);
}

在监听器中，你必须手动检查表示反序列化失败的 null 值：

@KafkaListener(id = "batch-deser", topics = "orders", containerFactory = "batchFactory")
public void listen(List<ConsumerRecord<String, Order>> records) {
    for (ConsumerRecord<String, Order> record : records) {
        if (record.value() == null) {
            // Deserialization failed - throw exception to send to DLT
            throw new BatchListenerFailedException("Deserialization failed", record);
        }
        process(record.value());
    }
}

当 DeadLetterPublishingRecoverer 将反序列化失败发布到 DLT 时：

未能反序列化的原始 byte[] 数据被恢复为记录值
异常信息（类名、消息、堆栈跟踪）被添加到标准 DLT 异常头
默认会移除原始 ErrorHandlingDeserializer 异常头（在恢复器上设置 setRetainExceptionHeader(true) 以保留它）

重试整个批

这现在是 DefaultErrorHandler 对批量监听器（监听器抛出的不是 BatchListenerFailedException 的异常时）的回退行为。

当一个批被重新投递时，无法保证该批具有相同数量的记录，和/或重新投递的记录顺序相同。因此，难以轻易地为批维护重试状态。FallbackBatchErrorHandler 采取以下方式。如果批量监听器抛出的不是 BatchListenerFailedException，则从内存中的记录批执行重试。为避免在延长的重试序列期间发生再均衡，错误处理器会暂停消费者，在每次重试休眠退避之前对其进行一次 poll，然后再次调用监听器。当/在重试用尽时，对批中的每条记录调用 ConsumerRecordRecoverer。如果恢复器抛出异常，或线程在休眠期间被中断，则该记录批将在下一次 poll 时重新投递。退出前，无论结果如何，消费者都会被恢复。

important

此机制不能与事务一起使用。

在等待 BackOff 间隔时，错误处理器会以短休眠循环直到达到期望延迟，同时检查容器是否已停止，使休眠能在 stop() 之后尽快退出，而不是造成延迟。

停止容器的错误处理器

如果监听器抛出异常，CommonContainerStoppingErrorHandler 会停止容器。对于记录监听器，当 AckMode 为 RECORD 时，已处理记录的偏移量被提交。对于记录监听器，当 AckMode 为任意手动值时，已确认记录的偏移量被提交。对于记录监听器，当 AckMode 为 BATCH 时，或对于批量监听器，容器重启时整个批会被重放。

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

委托错误处理器

CommonDelegatingErrorHandler 可以根据异常类型委托给不同的错误处理器。例如，你可能希望对大多数异常调用 DefaultErrorHandler，对其他异常调用 CommonContainerStoppingErrorHandler。

所有委托者必须共享相同的兼容属性（ackAfterHandle、seekAfterError 等）。

记录错误处理器

CommonLoggingErrorHandler 只是记录异常；对于记录监听器，上一次 poll 的剩余记录被传递给监听器。对于批量监听器，批中的所有记录都被记录。

对记录监听器和批量监听器使用不同的通用错误处理器

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

通用错误处理器汇总

DefaultErrorHandler
CommonContainerStoppingErrorHandler
CommonDelegatingErrorHandler
CommonLoggingErrorHandler
CommonMixedErrorHandler

旧版错误处理器及其替代者

旧版错误处理器	替代者
`LoggingErrorHandler`	`CommonLoggingErrorHandler`
`BatchLoggingErrorHandler`	`CommonLoggingErrorHandler`
`ConditionalDelegatingErrorHandler`	`DelegatingErrorHandler`
`ConditionalDelegatingBatchErrorHandler`	`DelegatingErrorHandler`
`ContainerStoppingErrorHandler`	`CommonContainerStoppingErrorHandler`
`ContainerStoppingBatchErrorHandler`	`CommonContainerStoppingErrorHandler`
`SeekToCurrentErrorHandler`	`DefaultErrorHandler`
`SeekToCurrentBatchErrorHandler`	无替代者，使用带无限 `BackOff` 的 `DefaultErrorHandler`。
`RecoveringBatchErrorHandler`	`DefaultErrorHandler`
`RetryingBatchErrorHandler`	无替代者，使用 `DefaultErrorHandler` 并抛出 `BatchListenerFailedException` 之外的异常。

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

请参阅 CommonErrorHandler 中的 JavaDocs。

要替换 ErrorHandler 或 ConsumerAwareErrorHandler 实现，你应实现 handleOne()，并让 seeksAfterHandle() 返回 false（默认）。你还应实现 handleOtherException() 以处理在记录处理范围之外发生的异常（例如消费者错误）。

要替换 RemainingRecordsErrorHandler 实现，你应实现 handleRemaining() 并覆盖 seeksAfterHandle() 返回 true（错误处理器必须执行必要的 seek）。你还应实现 handleOtherException()——以处理在记录处理范围之外发生的异常（例如消费者错误）。

要替换任意 BatchErrorHandler 实现，你应实现 handleBatch()。你还应实现 handleOtherException()——以处理在记录处理范围之外发生的异常（例如消费者错误）。

回滚后处理器

使用事务时，如果监听器抛出异常（并且存在的错误处理器也抛出异常），事务会回滚。默认情况下，任何未处理记录（包括失败记录）都会在下一次 poll 时被重新获取。这是通过在 DefaultAfterRollbackProcessor 中执行 seek 操作实现的。对于批量监听器，整个记录批会被重新处理（容器不知道批中哪条记录失败）。要修改此行为，你可以用自定义的 AfterRollbackProcessor 配置监听容器。例如，对于基于记录的监听器，你可能希望跟踪失败记录，并在若干次尝试后放弃它，也许通过将其发布到死信主题。

从 2.2 版本开始，DefaultAfterRollbackProcessor 现在可以恢复（跳过）一条不断失败的记录。默认情况下，在十次失败后，失败记录会被记录（以 ERROR 级别）。你可以用自定义恢复器（BiConsumer）和最大失败次数来配置该处理器。将 maxFailures 属性设置为负数会导致无限重试。以下示例配置了三次尝试后恢复：

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 版本开始，恢复现在可以恢复（跳过）不断失败的整个记录批。设置 ContainerProperties.setBatchRecoverAfterRollback(true) 以启用此特性。

important

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

另请参阅发布死信记录。

从 2.2.5 版本开始，DefaultAfterRollbackProcessor 可以在一个新事务中被调用（在失败事务回滚之后开启）。这样，如果你使用 DeadLetterPublishingRecoverer 发布失败记录，处理器会将被恢复记录在原始主题/分区中的偏移量发送到事务。要启用此特性，请在 DefaultAfterRollbackProcessor 上设置 commitRecovered 和 kafkaTemplate 属性。

important

如果恢复器失败（抛出异常），失败记录将被包含在 seek 中。从 2.5.5 版本开始，如果恢复器失败，默认会重置 BackOff，在再次尝试恢复之前，重新投递将再次经历各次退避。在更早版本中，BackOff 不会被重置，恢复会在下一次失败时再次尝试。要恢复此前行为，将处理器的 resetStateOnRecoveryFailure 属性设置为 false。

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

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

如果该函数返回 null，将使用处理器的默认 BackOff。

从 2.6.3 版本开始，将 resetStateOnExceptionChange 设置为 true，那么如果两次失败之间异常类型发生变化，重试序列将重新开始（包括在如此配置时选择新的 BackOff）。默认情况下，不考虑异常类型。

从 2.3.1 版本开始，与 DefaultErrorHandler 类似，DefaultAfterRollbackProcessor 将某些异常视为致命，对此类异常跳过重试；恢复器在第一次失败时被调用。默认被视为致命的异常有：

DeserializationException
MessageConversionException
ConversionException
MethodArgumentResolutionException
NoSuchMethodException
ClassCastException

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

你可以向不可重试类别添加更多异常类型，或者完全替换已分类异常的映射。更多信息请参阅 DefaultAfterRollbackProcessor.setClassifications() 的 Javadocs，以及 ExceptionMatcher。

以下示例向不可重试异常添加了 IllegalArgumentException：

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

另请参阅投递尝试头。

important

对于当前的 kafka-clients，容器无法检测 ProducerFencedException 是由再均衡引起，还是生产者的 transactional.id 因超时或过期被撤销。由于在大多数情况下它是由再均衡引起的，容器不会调用 AfterRollbackProcessor（因为我们已不再被分配这些分区，seek 它们并不合适）。如果你确保超时足够大以处理每个事务，并定期执行一个 "空" 事务（例如通过 ListenerContainerIdleEvent），就可以避免因超时和过期导致的围栏隔离。或者，你可以将 stopContainerWhenFenced 容器属性设置为 true，容器就会停止，从而避免记录丢失。你可以消费 ConsumerStoppedEvent 并检查 Reason 属性是否为 FENCED 以检测此情况。由于该事件还带有对容器的引用，你可以使用此事件重启容器。

从 2.7 版本开始，在等待 BackOff 间隔时，错误处理器会以短休眠循环直到达到期望延迟，同时检查容器是否已停止，使休眠能在 stop() 之后尽快退出，而不是造成延迟。

从 2.7 版本开始，处理器可以配置一个或多个 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 版本开始，当使用实现了 DeliveryAttemptAware 的 ErrorHandler 或 AfterRollbackProcessor 时，可以启用向记录添加 KafkaHeaders.DELIVERY_ATTEMPT 头（kafka_deliveryAttempt）。该头的值是一个从 1 开始递增的整数。当接收原始 ConsumerRecord<?, ?> 时，该整数位于一个 byte[4] 中。

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

当对 JsonKafkaHeaderMapper 或 SimpleKafkaHeaderMapper 使用 @KafkaListener 时，可以通过将 @Header(KafkaHeaders.DELIVERY_ATTEMPT) int delivery 作为参数添加到监听器方法来获取它。

要启用此头的填充，将容器属性 deliveryAttemptHeader 设置为 true。它默认被禁用，以避免为每条记录查找状态并添加头所带来的（微小）开销。

DefaultErrorHandler 和 DefaultAfterRollbackProcessor 都支持此特性。

批量监听器的投递尝试头

使用 BatchListener 处理 ConsumerRecord 时，KafkaHeaders.DELIVERY_ATTEMPT 头的存在方式可能与 SingleRecordListener 不同。

从 3.3 版本开始，如果你想在使用 BatchListener 时将 KafkaHeaders.DELIVERY_ATTEMPT 头注入 ConsumerRecord，请将 DeliveryAttemptAwareRetryListener 设为 ErrorHandler 中的 RetryListener。

请参阅以下代码。

final FixedBackOff fixedBackOff = new FixedBackOff(1, 10);
final DefaultErrorHandler errorHandler = new DefaultErrorHandler(fixedBackOff);
errorHandler.setRetryListeners(new DeliveryAttemptAwareRetryListener());

ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory);
factory.setCommonErrorHandler(errorHandler);

然后，每当一个批未能完成时，DeliveryAttemptAwareRetryListener 会向 ConsumerRecord 注入一个 KafkaHeaders.DELIVERY_ATTMPT 头。

监听器信息头

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

从 2.8.4 版本开始，你现在可以在监听容器上设置 listenerInfo 属性，或在 @KafkaListener 注解上设置 info 属性。然后，容器会把它以 KafkaListener.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) {
    ...
}

当在 RecordInterceptor 或 RecordFilterStrategy 实现中使用时，该头在消费者记录中是一个字节数组，使用 KafkaListenerAnnotationBeanPostProcessor 的 charSet 属性转换。

头映射器在从消费者记录创建 MessageHeaders 时也会转换为 String，并且从不在出站记录上映射此头。

对于 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) {
            ...
}

备注

如果批量监听器有一个过滤器，且过滤器导致空批，你需要向 @Header 参数添加 required = false，因为空批时 info 不可用。

如果你接收 List<Message<Thing>>，info 位于每个 Message<?> 的 KafkaHeaders.LISTENER_INFO 头中。

有关消费批的更多信息，请参阅批量监听器。

发布死信记录

当一条记录达到最大失败次数时，你可以用记录恢复器配置 DefaultErrorHandler 和 DefaultAfterRollbackProcessor。该框架提供了 DeadLetterPublishingRecoverer，它将失败消息发布到另一个主题。该恢复器需要一个 KafkaTemplate<Object, Object>，用于发送记录。你还可以选择用一个 BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition> 来配置它，该函数被调用来解析目标主题和分区。

important

默认情况下，死信记录被发送到名为 <originalTopic>-dlt 的主题（原始主题名加 -dlt 后缀），并发送到与原始记录相同的分区。因此，当你使用默认解析器时，死信主题 必须至少具有与原始主题相同数量的分区。

如果返回的 TopicPartition 的分区为负数，则该分区不会在 ProducerRecord 中设置，因此分区由 Kafka 选择。从 2.2.4 版本开始，任何 ListenerExecutionFailedException（例如在 @KafkaListener 方法中检测到异常时抛出的）都增强了 groupId 属性。这使目标解析器可以利用该信息以及 ConsumerRecord 中的信息来选择死信主题。

以下示例展示了如何装配自定义目标解析器：

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：异常类名（通常是 ListenerExecutionFailedException，但也可以是其他）。
KafkaHeaders.DLT_EXCEPTION_CAUSE_FQCN：异常 cause 类名（如果存在）（自 2.8 版本起）。
KafkaHeaders.DLT_EXCEPTION_STACKTRACE：异常堆栈跟踪。
KafkaHeaders.DLT_EXCEPTION_MESSAGE：异常消息。
KafkaHeaders.DLT_KEY_EXCEPTION_FQCN：异常类名（仅限键反序列化错误）。
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 版本起）。

键异常仅由 DeserializationException 引起，因此没有 DLT_KEY_EXCEPTION_CAUSE_FQCN。

有两种机制可用于添加更多头。

继承恢复器并重写 createProducerRecord()——调用 super.createProducerRecord() 并添加更多头。
提供一个 BiFunction 来接收消费者记录和异常，返回一个 Headers 对象；其中的头将被复制到最终生产者记录；另请参阅管理死信记录头。使用 setHeadersFunction() 设置该 BiFunction。

第二种更易实现，但第一种可获得更多信息，包括已组装好的标准头。

从 2.3 版本开始，与 ErrorHandlingDeserializer 一起使用时，发布者会在死信生产者记录中将记录 value() 恢复为未能反序列化的原始值。此前，value() 为 null，用户代码必须从消息头解码 DeserializationException。此外，你可以向发布者提供多个 KafkaTemplate；例如，当你想发布来自 DeserializationException 的 byte[]，以及为已成功反序列化的记录使用不同序列化器的值时，可能需要这样做。以下是一个示例，为发布者配置使用 String 和 byte[] 序列化器的 KafkaTemplate：

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

发布者使用映射键来定位适合即将发布的 value() 的模板。建议使用 LinkedHashMap，以便按键的顺序进行检查。

发布 null 值且存在多个模板时，恢复器会查找 Void 类的模板；如果不存在，则使用 values().iterator() 的第一个模板。

从 2.7 开始，你可以使用 setFailIfSendResultIsError 方法，以便在消息发布失败时抛出异常。你还可以用 setWaitForSendResultTimeout 设置验证发送方成功的超时时间。

important

如果恢复器失败（抛出异常），失败记录将被包含在 seek 中。从 2.5.5 版本开始，如果恢复器失败，默认会重置 BackOff，在再次尝试恢复之前，重新投递将再次经历各次退避。在更早版本中，BackOff 不会被重置，恢复会在下一次失败时再次尝试。要恢复此前行为，将错误处理器的 resetStateOnRecoveryFailure 属性设置为 false。

从 2.3 版本开始，该恢复器也可与 Kafka Streams 一起使用——更多信息请参阅从反序列化异常恢复。

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

如果入站记录相互依赖，但可能乱序到达，那么将失败记录重新发布到原始主题的尾部（若干次）可能会有用，而不是直接将其发送到死信主题。示例请参阅这个 Stack Overflow 问题。

以下错误处理器配置将恰好做到这一点：

@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 版本开始，恢复器会检查目标解析器选择的分区是否实际存在。如果该分区不存在，ProducerRecord 中的分区被设置为 null，让 KafkaProducer 选择分区。你可以通过将 verifyPartition 属性设置为 false 来禁用此检查。

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

管理死信记录头

参考上文发布死信记录，DeadLetterPublishingRecoverer 有两个属性用于管理那些已存在的头（例如在重新处理一条失败的死信记录时，包括使用非阻塞重试时）。

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

Apache Kafka 支持多个同名的头；要获取 "最新" 值，可以使用 headers.lastHeader(headerName)；要获得多个头的迭代器，使用 headers.headers(headerName).iterator()。

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

这两个属性存在的原因是：虽然你可能希望只保留最后一次异常信息，但你可能希望保留每条记录在每次失败时经过了哪些主题的历史。

appendOriginalHeaders 应用于所有名为 **ORIGINAL** 的头，而 stripPreviousExceptionHeaders 应用于所有名为 **EXCEPTION** 的头。

从 2.8.4 版本开始，你现在可以控制哪些标准头会被添加到输出记录。请参阅 enum HeadersToAdd，了解默认添加的（当前）10 个标准头的通用名称（这些不是实际头名，只是一个抽象；实际头名由 getHeaderNames() 方法设置，子类可以覆盖它）。

要排除头，使用 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 提供了多种 BackOff 实现。默认情况下，ExponentialBackOff 会无限重试；要在若干次重试尝试后放弃，需要计算 maxElapsedTime。自 2.7.3 版本起，Spring for Apache Kafka 提供了 ExponentialBackOffWithMaxRetries，它是一个接收 maxRetries 属性并自动计算 maxElapsedTime 的子类，稍微更方便一些。

@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​

使用死信主题的批量监听器错误处理​

使用 BatchListenerFailedException​

为批量监听器配置死信主题​

批量错误处理的工作方式​

对特定异常跳过重试​

偏移量提交行为​

批量错误处理器的转换错误​

批量监听器的反序列化错误​

重试整个批​

停止容器的错误处理器​

委托错误处理器​

记录错误处理器​

对记录监听器和批量监听器使用不同的通用错误处理器​

通用错误处理器汇总​

旧版错误处理器及其替代者​

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

回滚后处理器​

投递尝试头​

批量监听器的投递尝试头​

监听器信息头​

发布死信记录​

管理死信记录头​

ExponentialBackOffWithMaxRetries 实现​

监听器错误处理器

容器错误处理器

退避处理器

DefaultErrorHandler

使用死信主题的批量监听器错误处理

使用 BatchListenerFailedException

为批量监听器配置死信主题

批量错误处理的工作方式

对特定异常跳过重试

偏移量提交行为

批量错误处理器的转换错误

批量监听器的反序列化错误

重试整个批

停止容器的错误处理器

委托错误处理器

记录错误处理器

对记录监听器和批量监听器使用不同的通用错误处理器

通用错误处理器汇总

旧版错误处理器及其替代者

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

回滚后处理器

投递尝试头

批量监听器的投递尝试头

监听器信息头

发布死信记录

管理死信记录头

`ExponentialBackOffWithMaxRetries` 实现