Kafka 消费组丢失未提交的消息答案

【问题标题】：Kafka consumer group loses not commited messagesKafka 消费组丢失未提交的消息
【发布时间】：2018-03-19 10:10:23
【问题描述】：

我使用的消费者组只有一个消费者，只有一个经纪人（docker wurstmeister image）。在代码中决定是否提交偏移量 - 如果代码返回错误，则不提交消息。我需要确保系统不会丢失任何消息——即使这意味着永远重试相同的消息（现在；））。为了测试这一点，我创建了简单的处理程序，它不会在“错误”字符串作为消息发送到 kafka 的情况下提交偏移量。所有其他字符串都已提交。

kafka-console-producer --broker-list localhost:9092 --topic test
>this will be commited

正在运行

kafka-run-class kafka.admin.ConsumerGroupCommand --bootstrap-server localhost:9092 --group michalgrupa --describe

TOPIC                          PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG        CONSUMER-ID                                       HOST                           CLIENT-ID
test                           0          13              13              0

没关系，没有滞后。现在我们传递 'error' 字符串来假装发生了不好的事情并且没有提交消息：

TOPIC                          PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG        CONSUMER-ID                                       HOST                           CLIENT-ID
test                           0          13              14              1

当前偏移量保持在正确位置 + 有 1 条滞后消息。现在，如果我们再次传递正确的消息，偏移量将移动到 15：

TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG test 0 15 15

并且第 14 号消息将不会再被拾取。这是默认行为吗？我是否需要手动跟踪最后一个偏移量并通过 it+1 加载消息？我已将提交间隔设置为 0，希望不使用任何 auto.commit 机制。

获取/提交代码：

go func() {
    for {
        ctx := context.Background()

        m, err := mr.brokerReader.FetchMessage(ctx)
        if err != nil {
            break
        }

        if err := msgFunc(m); err != nil {
            log.Errorf("# messaging # cannot commit a message: %v", err)
            continue
        }

        // commit message if no error
        if err := mr.brokerReader.CommitMessages(ctx, m); err != nil {
            // should we do something else to just logging not committed message?
            log.Errorf("cannot commit message [%s] %v/%v: %s = %s; with error: %v", m.Topic, m.Partition, m.Offset, string(m.Key), string(m.Value), err)
        }
    }
}()

阅读器配置：

kafkaReader := kafka.NewReader(kafka.ReaderConfig{
Brokers:         brokers,
GroupID:         groupID,
Topic:           topic,
CommitInterval:  0,
MinBytes:        10e3,
MaxBytes:        10e6,
})

使用的库：https://github.com/segmentio/kafka-go

【问题讨论】：

标签： go apache-kafka

【解决方案1】：

在 kafka 中，您只需提交偏移量而不是单个消息。如果我正确理解您的代码（不是开发人员）。您只需在遇到无效消息后继续。如果在无效消息中出现有效消息后，您将再次提交偏移量 - 我想这不是您的意图。

只是为了明确提交或提交偏移量的含义：您的消费者组会将偏移量存储到专用的内部 kafka 主题（或在 zookeeper 上的较旧 kafka 版本上）。偏移量可以识别主题中的单个位置（或者更准确地说是在给定主题的分区上）。这意味着您只能以线性方式使用主题。

在这里你可以看到在 kafka-consumer 端发生了什么：

您正在从（很可能是多个）堆栈消息中消费。您在此主题/分区提交位置（a.k.a 偏移量）。因此，您可以不说我想再次接收特定消息。您可以做的是在收到无效消息后停止消费。在这种情况下，您的问题将是：如何摆脱此消息。从 kafka 主题中删除一条消息很棘手。一种常见的模式是将此消息写入某种死信主题并由不同的消费者处理。

希望这能让你更清楚一点。

【讨论】：

但我想它应该获取完全相同的消息，因为它没有提交，但它会按照你说的那样继续。如何强制轮询相同的消息？我相信它更多与 kafka 相关的知识不是 go/library 本身。
完全同意：这不是一个 go 问题，而是一个关于 kafka 实际工作方式的问题。我更新了我的答案，也许可以更好地解释它
我没有得到的是提交偏移部分。你写了“你提交了这个主题/分区的位置（a.k.a offset）”，这就是问题 - 我不提交它。我不提交偏移量:) 它会在下一条有效消息上继续前进。
再次不是 Go 开发者。但我会解释这部分代码：if err := msgFunc(m); err != nil { log.Errorf("# messaging # cannot commit a message: %v", err) continue } 如下。当您遇到错误时，请继续下一条消息。如果这个有效，则提交偏移量。因此，假设我们在一个主题上有 3 条消息（为了让一个分区保持简单）。消息 2 无效。您消费者 m1 提交偏移量，消耗 m2 不提交偏移量，然后消耗 m3 并再次提交偏移量。所以你提交的偏移量是 3
没错，我们终于在同一页上 :) 所以提交偏移量 m3 忽略了 m2 没有提交。为了解决这个问题，我相信我必须做一些手动检查，在提交之前我只在位置 offset+1 上。

【解决方案2】：

在这里理解消费者抵消的概念是有意义的。消费者应用运行时，无论commit/uncommit offset如何，都会将消费消息的偏移量存储在内存中，如果重启消费者应用程序，它将检索'CURRENT-OFFSET'的偏移量以继续消费。

【讨论】：

【解决方案3】：

您的 Kafka 使用者似乎已设置为自动提交偏移量（这是默认设置）。
如果是这样，这可能就是您的应用跳过错误消息的原因 - 尽管您跳过了 CommitMessages 调用，但提交是在后台线程上执行的
请查看文档中的enable.auto.commit 属性规范：https://kafka.apache.org/documentation/#newconsumerconfigs

【讨论】：