使用 Confluent.Kafka 消费者库的 Kafka 消息密钥为空答案

【问题标题】：Kafka message key is null using Confluent.Kafka consumer library使用 Confluent.Kafka 消费者库的 Kafka 消息密钥为空
【发布时间】：2021-07-22 01:33:32
【问题描述】：

我正在使用 Debezium 对我的数据库执行 CDC 以在 Kafka 中创建消息。使用 kafdrop 和 OffsetExplorer 等工具，我可以看到消息 key 和 value。但是，在使用 Confluent.Kafka 库的 .NET 框架应用程序中，当我使用消息时，消息 key 始终为空。

如何使用 Confluent.Kafka 库检索消息 key？

这是VS项目的代码：

using System;
using Confluent.Kafka;
using System.Threading;

namespace KafkaConsumer
{
    class Program
    {
        static void Main(string[] args)
        {
            var config = new ConsumerConfig
            {
                BootstrapServers = "kakfa:9092,localhost:9093",
                GroupId = "foo",
                AutoOffsetReset = AutoOffsetReset.Earliest,
                
            };

            var topics = "CAC_connector.dbo.sessionLogs";
            bool cancelled = false;

            // Define the cancellation token.
            CancellationTokenSource source = new CancellationTokenSource();
            CancellationToken cancellationToken = source.Token;

            using (var consumer = new ConsumerBuilder<Ignore, string>(config).Build())
            {
                consumer.Subscribe(topics);

                while (!cancelled)
                {
                    var cr = consumer.Consume(cancellationToken);
                    Console.WriteLine($"Consumed record with key {cr.Message.Key} and value {cr.Message.Value.Substring(0, 96)}");
                }

                consumer.Close();
            }
        }
    }
}

干杯，凯莉

【问题讨论】：

您希望ConsumerBuilder<Ignore 做什么？
从另一个示例复制代码时，我没有注意到这一点。我正在寻找配置设置，但不了解构造函数参数。谢谢你向我指出这一点。当更改为 ConsumerBuilder<string 时，它正在按我的预期工作。再次感谢。

标签： c# apache-kafka kafka-consumer-api confluent-kafka-dotnet

【解决方案1】：

正如 OneCricketeer 指出的那样，我最初构建 ConsumerBuilder 是为了忽略密钥。当我将构造函数更改为 ConsumerBuilder<string, string> 时，密钥包含在消息中。

【讨论】：