spring-kafka 的 Kafka 死信队列 (DLQ)

Question

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

所以消耗的Kafka记录是：

处理成功，
处理失败，发送到DLQ主题，
处理失败，没有发送到DLQ主题（由于意外问题）所以会被监听者再次消费

我尝试使用 ErrorHandler 的自定义实现创建侦听器容器，发送记录无法使用 KafkaTemplate 处理到 DLQ 主题。使用禁用的 auto-commit 和 RECORD 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);
        }
    }
}

如果消费者轮询 returns 3 条记录 (1, 2, 3) 并且无法处理第二条记录，则这样：