gpt4 book ai didi

apache-kafka - 消息处理失败后Kafka消费者恢复

转载 作者:行者123 更新时间:2023-12-04 04:48:52 24 4
gpt4 key购买 nike

我正在我的一个项目中使用简单的 kafka 消费者,我想要的逻辑是当消费者无法处理某些消息时,它将提交最后正确处理的消息,然后在下一次轮询时,它将从失败的消息继续。

我尝试使用以下代码手动提交每条消息:

public void fetchMessages() {
ConsumerRecords<String, MyObject> messages = kafkaConsumer.poll(10000);
for (ConsumerRecord message : messages) {
logger.info("Reading kafka message, topic ["+kafkaTopic+"], partition ["+message.partition()+"], offset ["+message.offset()+"]");
try {
MyObject myObject = (MyObject) message.value();
logger.info("Handling message," + myObject);
handleMessage(myObject);
commitMessage(message);
} catch (Exception e) {
logger.error("Error handling message"); throw e;
}
}
}


private void commitMessage(ConsumerRecord message) {
long nextOffset = message.offset() + 1;

TopicPartition topicPartition = new TopicPartition(kafkaTopic,message.partition());
OffsetAndMetadata offsetAndMetadata = new OffsetAndMetadata(nextOffset);

Map<TopicPartition,OffsetAndMetadata> offsetAndMetadataMap = new HashMap<>();
offsetAndMetadataMap.put(topicPartition,offsetAndMetadata);

logger.info("Commiting processed kafka message, topic ["+kafkaTopic+"], partition ["+message.partition()+"], next offset ["+nextOffset+"]");
kafkaConsumer.commitSync(offsetAndMetadataMap);
}

但是例如,当我获取 3 条消息时,每条消息都来自不同的分区,我成功处理了第一条消息,然后未能处理第二条消息,我只是退出 ConsumerRecord s for 循环,我希望在下一个 poll 中获得我尚未提交的相同 2 条消息迭代。相反,消费者只是继续接收新消息,永远不会返回失败的消息。

也试过申请 seek在失败的消息上,然后退出循环,但它正在 1 个分区上工作,并且在许多分区上都不起作用。
kafkaConsumer.seek(new TopicPartition(kafkaTopic,message.partition()),message.offset());    

一些细节:
  • 主题有 12 个分区
  • 所有分区的一个使用者
  • 消费者每分钟执行一次轮询循环
  • enable.auto.commit: false

  • 我的代码或逻辑有什么问题?

    最佳答案

    我发现了 seek 是如何工作的,在失败的消息中,我必须为当前使用者的所有分区寻找所有偏移量。

    private void seekAllPartitions() {
    logger.info("Processing of some kafka message was failed, seeking all partitions to last committed");
    List<PartitionInfo> partitionInfos = kafkaConsumer.partitionsFor(kafkaTopic);
    for (PartitionInfo partitionInfo : partitionInfos) {
    TopicPartition topicPartition = new TopicPartition(kafkaTopic, partitionInfo.partition());
    OffsetAndMetadata committedForPartition = kafkaConsumer.committed(topicPartition);
    if (committedForPartition != null) {
    kafkaConsumer.seek(topicPartition,committedForPartition.offset());
    }
    }
    }

    当某个分区上某个消费者组的最后一个偏移量尚未设置时,需要对 commitedForPartition 进行空检查(未知)

    关于apache-kafka - 消息处理失败后Kafka消费者恢复,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/40147309/

    24 4 0
    Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
    广告合作:1813099741@qq.com 6ren.com