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

各位小伙伴们，大家好呀！看看今天我又给各位带来了什么文章？本文标题是《有时候，当消费者重新启动时，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学习网公众号下多多回复交流；文中若有不正之处，也希望回复留言以告知！