在Kafka用户中重试用尽时如何设置确认 [英] How to set acknowledgement in case when retry gets exhausted in Kafka consumer

问题描述

我有一个重试5次的Kafka使用者，并且我将Spring Kafka与重试模板一起使用.现在，如果所有重试都失败了，那么在这种情况下如何确认工作.另外，如果我将确认模式设置为手动，那么如何确认这些消息

I have a Kafka consumer which retry 5 time and I am using Spring Kafka with retry template . Now if all retry are failed then how to does acknowledge work in that case . Also if i have set acknowledge mode to manually then how to acknowledge those message

消费者

@Bean("kafkaListenerContainerFactory")
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(RetryTemplate retryTemplate) {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
    factory.setRetryTemplate(retryTemplate);
    factory.setRecoveryCallback(context -> {
        log.error("Maximum retry policy has been reached {}", context.getAttribute("record"));
        Acknowledgment ack = (Acknowledgment) context.getAttribute(RetryingMessageListenerAdapter.CONTEXT_ACKNOWLEDGMENT);
        ack.acknowledge();
        return null;
    });
    factory.setConcurrency(Integer.parseInt(kafkaConcurrency));
    return factory;
}

Kafka监听器

@KafkaListener(topics = "${kafka.topic.json}", containerFactory = "kafkaListenerContainerFactory")
public void recieveSegmentService(String KafkaPayload, Acknowledgment acknowledgment) throws Exception {
    KafkaSegmentTrigger kafkaSegmentTrigger;
    kafkaSegmentTrigger = TransformUtil.fromJson(KafkaPayload, KafkaSegmentTrigger.class);
    log.info("Trigger recieved from segment service {}", kafkaSegmentTrigger);
    processMessage(kafkaSegmentTrigger);
    acknowledgment.acknowledge();
}

推荐答案

如果您有一个RecoveryCallback可以在重试用尽时处理记录，并且您没有使用手动方式，则容器将提交偏移量.

If you have a RecoveryCallback that handles the record when retries are exhausted, and you are NOT using manual acks, the container will commit the offset.

如果您使用的是MANUAL ack，则可以在恢复回调中提交偏移量.

If you are using MANUAL acks, you can commit the offset in the recovery callback.

传递给回调的上下文对象具有属性RetryingMessageListenerAdapter.CONTEXT_ACKNOWLEDGMENT(确认").

The context object passed to the callback has an attribute RetryingMessageListenerAdapter.CONTEXT_ACKNOWLEDGMENT ('acknowledgment').

它也有CONTEXT_CONSUMER和CONTEXT_RECORD.

这篇关于在Kafka用户中重试用尽时如何设置确认的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！