使用 Spring Kafka 批处理消费者进行非阻塞重试答案

【问题标题】：Non-blocking retries with Spring Kafka batch consumer使用 Spring Kafka 批处理消费者进行非阻塞重试
【发布时间】：2022-01-12 03:34:38
【问题描述】：

我正在使用 spring-kafka 2.8.0，我正在尝试为 batch kafka consumer 实现 non-blocking retries。这是我的配置和消费者：

@Configuration
public class KafkaConfig {
    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, GenericRecord>> 
            batchListenerFactory(ConsumerFactory<Object, Object> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, GenericRecord> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.setBatchListener(true);
        return factory;
    }
}

@Component
public class MyConsumer {

    @KafkaListener(
            topics = "my-topic",
            containerFactory = "batchListenerFactory"
    )
    @RetryableTopic(
            backoff = @Backoff(delay = 1000, multiplier = 2.0),
            attempts = "4",
            topicSuffixingStrategy = SUFFIX_WITH_INDEX_VALUE,
            autoCreateTopics = "false"
    )
    public void consume(List<ConsumerRecord<String, GenericRecord>> messages) {
        // do some stuff
    }

}

但是在 sturtup 上，我遇到了以下异常： java.lang.IllegalArgumentException: The provided class BatchMessagingMessageListenerAdapter is not assignable from AcknowledgingConsumerAwareMessageListener

我的问题是：

有没有办法将批量消费者与@RetryableTopic结合起来？
还有其他方法可以为批量消费者实现非阻塞重试吗？是否可以为此使用RetryTemplate？

【问题讨论】：

标签： java spring spring-boot apache-kafka spring-kafka

【解决方案1】：

批处理侦听器不支持@RetryableTopic。

RecoveringBatchErrorHandler（DefaultErrorHandler 用于 2.8 及更高版本）支持将批处理中的失败记录发送到死信主题，在侦听器的帮助下抛出 BatchListenerFailedException 指示哪个记录失败。

然后，您必须针对该主题实现自己的侦听器。

【讨论】：