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

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条消息不会再收到了。是违约行为吗？我是否需要跟踪最后一个偏移量并通过它+1手动加载消息？我已经将commit interval设置为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