卡夫卡消费者丢失消息答案

【问题标题】：Kafka consumer missing messages卡夫卡消费者丢失消息
【发布时间】：2021-01-26 13:14:54
【问题描述】：

在 2 个分区上有名为“WebMessages”的 Kafka 主题。

我们在同一台服务器上有两个消费者组，但在 IIS 上的不同站点。

其中一个消费者无法接收消息。另一个错过了大部分消息。

当我在本地计算机上编写简单的消费者时，我也错过了一些消息。知道出了什么问题吗？

这里是生产者代码：

_producerConfig = new ProducerConfig {
                    BootstrapServers = _addressWithPort,
                    Acks = Acks.All
                };

using (var p = new ProducerBuilder<string, string>(_producerConfig).Build())
                {
                    p.ProduceAsync(_topicName, new Message<string, string>
                    {
                        Key = ldtoKafkaMessage.Key,
                        Value = ldtoKafkaMessage.Message
                    }).ContinueWith(task =>
                    {
                        if (task.IsFaulted)
                        {
                            TraceController.TraceError(Common.Enums.TraceEventCategories.X, "Key|Message", ldtoKafkaMessage.Key + " " + ldtoKafkaMessage.Message + " " + task.Exception.Message + " " + task.Exception.InnerException+ " " + task.Exception.StackTrace);
                        }
                        else
                        {
                            TraceController.TraceInformation(Common.Enums.TraceEventCategories.X, "Key|Message|Result", ldtoKafkaMessage.Key + " " + ldtoKafkaMessage.Message + " " + "Success");
                        }
                    });
                }

所以我确保我向生产者发送了消息。

这是消费者代码。

     _consumerConfig = new ConsumerConfig
        {
            BootstrapServers = _addressWithPort,
            AutoOffsetReset = AutoOffsetReset.Earliest,
            GroupId = consumerGroupId
        };

  using (var c = new ConsumerBuilder<string, string>(_consumerConfig).Build())
            {
                c.Subscribe(_topicName);

                CancellationTokenSource cts = new CancellationTokenSource();
                try
                {
                    while (!cts.IsCancellationRequested)
                    {
                        try
                        {
                            var cr = c.Consume(cts.Token);
                            LdtoKafkaMessage.Key = cr.Key;
                            LdtoKafkaMessage.Message = cr.Value;
                            this.OnMessageChanged();
                        }
                        catch (ConsumeException e)
                        {

                            Console.WriteLine($"Error occured: {e.Error.Reason}");
                        }
                    }
                }
                catch (OperationCanceledException)
                {
                    // Ensure the consumer leaves the group cleanly and final offsets are committed.
                    c.Close();
                }
            }

【问题讨论】：

要缩小问题范围，您应该首先检查是否所有消息都写入了您的主题。或许是制作人的错？您可以从kafka.apache.org/downloads 下载kafka 工具并使用./kafka-consumer-groups.bat --bootstrap-server BOOTSTRAP_SERVER --describe --group CONSUMER_GROUP 检查偏移量
我们查看了该主题。看起来丢失的消息也不在主题中。问题看起来与生产者有关。谢谢@rytisk
相关：stackoverflow.com/questions/63950185/…

标签： c# apache-kafka kafka-consumer-api kafka-producer-api

【解决方案1】：

您的某些消息可能不会生成，因为您处置了生产者并且不等待它完成。您可以通过在 ProduceAsync 方法上使用 await 关键字或在处置生产者之前调用 Flush() 来确保消息传递。

【讨论】：