登录
首页 >  Golang >  Go问答

有时候,当消费者重新启动时,Kafka 偏移量会重置

来源:stackoverflow

时间:2024-02-18 14:48:25 454浏览 收藏

各位小伙伴们,大家好呀!看看今天我又给各位带来了什么文章?本文标题《有时候,当消费者重新启动时,Kafka 偏移量会重置》,很明显是关于Golang的文章哈哈哈,其中内容主要会涉及到等等,如果能帮到你,觉得很不错的话,欢迎各位多多点评和分享!

问题内容

我有一个 kafka 主题,有 4 个分区和 4 个具有相同消费者组 id 的消费者。 所有消费者每天晚上都会重新启动,第一次是凌晨 4:00,第二次是凌晨 4:15,其他时间是 4:30 和 4:45(这是我们管理员的规则,我对此无能为力)。 有时某些分区的偏移量会重置,并且消费者在重新启动后会读取以偏移量 0 开头的所有消息。有时不会发生这种情况。有什么想法问题出在哪里吗? 消费者在 go 中。

卡夫卡配置:

conf := kafka.readerconfig{
        brokers:  cfg.kafka.bootstrapservers,
        groupid:  cfg.kafka.consumergroup,
        topic:    cfg.kafka.topic,
        minbytes: 10e3, // 10kb
        maxbytes: 32e6, // 10mb
    }

阅读消息:

for {
        m, err := reader.ReadMessage(context.Background())
        if err != nil {
            go log.Error(componentName, actionName, -123753, err.Error(), "Ошибка чтения сообщения с кафки", "", "", string(m.Value), nil)
            continue
        }
        log.Warn(componentName, actionName, "Обработка сообщения", "START", "", "", map[string]interface{}{"offset": m.Offset, "partition": m.Partition}, nil)
        var msg model.KafkaMessage

        err = json.Unmarshal(m.Value, &msg)
        if err != nil {
            go log.Error(componentName, actionName, -123754, err.Error(), "Ошибка парсинга сообщения", "", "", string(m.Value), nil)
            continue
        }

        _service := service.NewService(log, cfg)
        currentNumAttempts := 0 
        maxNumAttempt := 3      

        for currentNumAttempts < maxNumAttempt {
            currentNumAttempts++
            err = _service.UpdateNumberInColvir(msg)
            if err != nil {
                continue
            }
            break
        }

    }

正确答案


根据您的评论And cleanup.policy是DELETE以下情况可能会发生。

__consumer_offsets 主题跟踪消费者组的每个主题分区的最后提交偏移量。

cleanup.policy = delete 表示您的 __consumer_offsets 主题分区日志在 retention.ms 之后删除。

因此,您的消费者在 retention.ms 期间没有消费任何消息,您的 __consumer_offset 主题将不会更新,并且偏移主题中的所有消息都将被删除。那么 __consumer_offsets 主题中没有任何跟踪到您的消费者组。因此,Kafka认为它是作为一个新的消费者组到主题分区并从头开始读取消息。

注意 - cleanup.policy 应该是 __consumer.offsets 主题的 compact

今天关于《有时候,当消费者重新启动时,Kafka 偏移量会重置》的内容介绍就到此结束,如果有什么疑问或者建议,可以在golang学习网公众号下多多回复交流;文中若有不正之处,也希望回复留言以告知!

声明:本文转载于:stackoverflow 如有侵犯,请联系study_golang@163.com删除
相关阅读
更多>
最新阅读
更多>
课程推荐
更多>