Dead Letter queue (DLQ) for Kafka with spring-kafka

在 Spring Boot 2.0 应用程序中实现死信队列 (DLQ) 概念的最佳方法是什么,使用 spring-kafka 2.1.x 将所有失败的消息发送到 @KafkaListener某个预定义的 Kafka DLQ 主题,而不会丢失单个消息?

所以消耗的卡夫卡记录是:

  1. 已成功处理,
  2. 处理失败,并发送到 DLQ 主题,
  3. 处理失败,未发送到DLQ主题(由于意外问题),因此将被监听器再次消耗。

我尝试使用错误处理程序的自定义实现创建侦听器容器,该容器使用 KafkaTemplate 将记录处理失败到 DLQ 主题。使用禁用的自动提交和记录 AckMode。

spring.kafka.enable-auto-ack=false
spring.kafka.listener.ack-mode=RECORD

@Configuration
public class KafkaConfig {
    @Bean
    ConcurrentKafkaListenerContainerFactory<Integer, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory = ...
        ...
        factory.getContainerProperties().setErrorHandler(dlqErrorHandler);
        return factory;
    }
}

@Component
public class DlqErrorHandler implements ErrorHandler {

    @Autowired
    private KafkaTemplate<Object, Object> kafkaTemplate;

    @Value("${dlqTopic}")
    private String dlqTopic;

    @Override
    public void handle(Exception thrownException, ConsumerRecord<?, ?> record) {
        log.error("Error, sending to DLQ...");
        kafkaTemplate.send(dlqTopic, record.key(), record.value());
    }
}

似乎此实现并不能保证项目 #3。如果在DlqErrorHandler记录中引发异常,则监听器不会再次使用。

使用事务侦听器容器是否有帮助?

factory.getContainerProperties().setTransactionManager(kafkaTransactionManager);

有没有方便的方法来使用Spring Kafka实现DLQ概念?

更新 2018/03/28

多亏了Gary Russell的回答,我能够通过实现DlqErrorHandler来实现所需的行为,如下所示

@Configuration
public class KafkaConfig {
    @Bean
    ConcurrentKafkaListenerContainerFactory<Integer, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory = ...
        ...
        factory.getContainerProperties().setAckOnError(false);
        factory.getContainerProperties().setErrorHandler(dlqErrorHandler);
        return factory;
    }
}

@Component
public class DlqErrorHandler implements ContainerAwareErrorHandler {
    ...
    @Override
    public void handle(Exception thrownException, list<ConsumerRecord<?, ?> records, Consumer<?, ?> consumer, MessageListenerContainer container) {
        Consumerrecord<?, ? record = records.get(0);
        try {
            kafkaTemplate.send("dlqTopic", record.key, record.value());
            consumer.seek(new TopicPartition(record.topic(), record.partition()), record.offset() + 1);
            // Other records may be from other partitions, so seek to current offset for other partitions too
            // ...
        } catch (Exception e) {
            consumer.seek(new TopicPartition(record.topic(), record.partition()), record.offset());
            // Other records may be from other partitions, so seek to current offset for other partitions too
            // ...
            throw new KafkaException("Seek to current after exception", thrownException);
        }
    }
}

这样,如果消费者轮询返回 3 条记录(1、2、3),而第 2 条记录无法处理:

  • 1 将被处理
  • 2 将无法处理并发送到 DLQ
  • 3 多亏了消费者寻求录音.offset() + 1,它将被传送给听众

如果发送到 DLQ 失败,使用者会向 record.offset() 寻求记录,并且该记录将被重新传递给侦听器(并且发送到 DLQ 可能会被停用)。

更新 2021/04/30

自Spring Kafka 2.7.0以来,本机支持非阻塞重试和死信主题

请参阅示例:https://github.com/evgeniy-khist/spring-kafka-non-blocking-retries-and-dlt

重试通常应为非阻塞(在单独的主题中完成)并延迟:

  • 不干扰实时流量;
  • 不增加呼叫数量,本质上是垃圾邮件不良请求;
  • 用于可观察性(获取重试次数和其他元数据上的数字)。使用 Kafka 实现非阻塞重试和 DLT 功能通常需要设置额外的主题并创建和配置相应的侦听器。Kafka non-blocking retries and DLT

答案 1

请参阅SeekToCurrentErrorHandler

发生异常时,它会查找使用者,以便在下一次轮询时重新传递所有未处理的记录。

您可以使用相同的技术(例如子类)写入 DLQ,并在 DLQ 写入失败时查找当前偏移量(以及其他未处理的偏移量),并在 DLQ 写入成功时仅查找剩余记录。

编辑

这是在发布此答案几个月后添加的。DeadLetterPublishingRecoverer

https://docs.spring.io/spring-kafka/docs/current/reference/html/#dead-letters


答案 2

推荐