Kafka的死信队列（DLQ）

Question

什么是最好的方法来实现死信队列（DLQ）的概念在Spring Boot 2.0应用程序中使用sping-kafka 2.1. x有所有的消息被处理失败的@KafkaListener方法的一些bean发送到一些预定义的Kafka DLQ主题并且不丢失一条信息？

因此，Kafka的记录是：

1. 已成功处理，
2. 处理失败，并发送到DLQ主题，
3. 处理失败，未发送到DLQ主题（由于意外问题），因此将再次被侦听器使用

我尝试使用ErrorHandler的自定义实现创建侦听器容器，使用KafkaTemplate将无法处理的记录发送到DLQ主题。使用禁用的自动提交和记录确认模式。

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），第二条记录无法处理：

• 1将被处理
• 2将处理失败并发送到DLQ
• 3感谢消费者寻求record.offset（）1，它将被传递给听众

如果发送到DLQ失败，消费者寻求记录。offset（）和记录将被重新发送到侦听器（发送到DLQ可能会失效）。

更新2021/04/30

自Spring以来，Kafka2.7.0无阻塞重试和死信主题在本机上得到支持。

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

重试通常应该是非阻塞的（在单独的主题中完成）并延迟：

• 不干扰实时通信

Answer 1

请参阅seekToCurInterrorHandler。

当出现异常时，它会寻找消费者，以便在下一次投票时重新提交所有未处理的记录。

如果DLQ写入失败，您可以使用相同的技术（例如子类）写入DLQ，并查找当前偏移量（和其他未处理的偏移量），如果DLQ写入成功，则只查找剩余记录。

编辑

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

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