spring-kafka 消费者批处理错误处理与 spring boot 版本 2.3.7答案

【问题标题】：spring-kafka consumer batch error handling with spring boot version 2.3.7spring-kafka 消费者批处理错误处理与 spring boot 版本 2.3.7
【发布时间】：2021-05-06 07:25:55
【问题描述】：

我正在尝试执行 spring kafka 批处理错误处理。首先我有几个问题。

侦听器和容器错误处理程序有什么区别？这两类错误有哪些？
您能否帮助一些示例以更好地理解？

这是我们的设计：

每隔一定时间轮询一次
以批处理模式消费消息
根据key推送到本地缓存（应用缓存）（避免重复事件）
批处理完成后，将所有值一个一个地推送到另一个主题。
操作 3 完成后清除缓存并手动确认偏移量。

这是我的错误处理计划：

public ConcurrentKafkaListenerContainerFactory<String, String> myListenerPartitionContainerFactory(String groupId) {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory(groupId));
        factory.setConcurrency(partionCount);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        factory.getContainerProperties().setIdleBetweenPolls(pollInterval);
        factory.setBatchListener(true);

        return factory;
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> myPartitionsListenerContainerFactory() 
    {
        return myListenerPartitionContainerFactory(groupIdPO);
    }


@Bean
public RecoveringBatchErrorHandler(KafkaTemplate<String, String> errorKafkaTemplate) {
    DeadLetterPublishingRecoverer recoverer =
            new DeadLetterPublishingRecoverer(errorKakfaTemplate);
    RecoveringBatchErrorHandler errorHandler =
            new RecoveringBatchErrorHandler(recoverer, new FixedBackOff(2L, 5000)); // push error event to the error topic
}


@KafkaListener(id = "mylistener", topics = "someTopic", containerFactory = "myPartitionsListenerContainerFactory"))
public void listen(List<ConsumerRecord<String, String>> records, @Header(KafkaHeaders.MESSAGE_KEY) String key, Acknowledgement ack) {
    Map hashmap = new Hashmap<>();
    records.forEach(record -> {
        try {
            //key will be formed based on the input record - it will be id.
            hashmap.put(key, record);  
        }
        catch (Exception e) {
            throw new BatchListenerFailedException("Failed to process", record);
        }
         
    });
    // Once success each messages to another topic.
    try {
      hashmap.forEach( (key,value) -> {  push to another topic })
      hashmap.clear();
      ack.acknowledge();
    } catch(Exception ex) {
        //handle producer exceptions
    }
}

方向好还是需要改进？以及需要实现什么类型的容器和侦听器处理程序？

@Gary Russell.. 你能帮忙吗？

【问题讨论】：

标签： spring-boot spring-kafka

【解决方案1】：

侦听器错误处理程序适用于错误处理程序可以向发送者返回有意义的回复的请求/回复情况。

你需要抛出一个异常来触发容器错误处理程序，并且你需要知道在原始批次的索引中告诉它哪条记录失败了。

如果您正在使用类似的手动确认，您可以使用 nack() 方法来指示哪个记录失败（并且在这种情况下不要抛出异常）。

【讨论】：