.NET使用Confluent.Kafka消费Debezium CDC消息时key为null如何解决

你当前代码中ConsumerBuilder的第一个泛型参数设置为Ignore，该参数定义消息Key的解析类型，Ignore会让消费者直接忽略消息Key的读取，因此你拿到的cr.Message.Key永远为null。

首先确认Debezium生成的Kafka消息Key的序列化格式，针对两种常见场景分别处理：

场景1：Key为JSON/纯字符串格式（Debezium默认未开启Avro序列化的场景） #

直接修改ConsumerBuilder的泛型参数，将Key类型改为string即可，修改后的代码如下：

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

namespace KafkaConsumer
{
    class Program
    {
        static void Main(string[] args)
        {
            var config = new ConsumerConfig
            {
                BootstrapServers = "kafka:9092,localhost:9093", // 原配置中kafka拼写错误已修正
                GroupId = "foo",
                AutoOffsetReset = AutoOffsetReset.Earliest,
            };

            var topics = "CAC_connector.dbo.sessionLogs";
            CancellationTokenSource source = new CancellationTokenSource();
            CancellationToken cancellationToken = source.Token;

            // 仅修改此处泛型参数，将Ignore改为string
            using (var consumer = new ConsumerBuilder<string, string>(config).Build())
            {
                consumer.Subscribe(topics);

                while (!cancellationToken.IsCancellationRequested)
                {
                    var cr = consumer.Consume(cancellationToken);
                    Console.WriteLine($"Consumed record with key {cr.Message.Key} and value {cr.Message.Value.Substring(0, 96)}");
                }
                // 新版本Confluent.Kafka无需手动调用Close，Dispose会自动释放资源
            }
        }
    }
}

场景2：Key为Avro序列化格式 #

如果Debezium配置了Avro序列化写入Schema Registry，需要额外做以下配置：

• 安装NuGet包：Confluent.Kafka.Avro、Confluent.SchemaRegistry.Serdes.Avro
• 在配置中新增Schema Registry地址
• 给Key和Value配置Avro反序列化器
  示例配置代码：

var config = new ConsumerConfig
{
    BootstrapServers = "kafka:9092,localhost:9093",
    GroupId = "foo",
    AutoOffsetReset = AutoOffsetReset.Earliest,
};
var schemaRegistryConfig = new SchemaRegistryConfig
{
    Url = "你的Schema Registry地址"
};

using var schemaRegistry = new CachedSchemaRegistryClient(schemaRegistryConfig);
using var consumer = new ConsumerBuilder<你的Key强类型, 你的Value强类型>(config)
    .SetKeyDeserializer(new AvroDeserializer<你的Key强类型>(schemaRegistry).AsSyncOverAsync())
    .SetValueDeserializer(new AvroDeserializer<你的Value强类型>(schemaRegistry).AsSyncOverAsync())
    .Build();

如果Debezium的Key是自定义序列化格式，可以自行实现IDeserializer<T>接口传入ConsumerBuilder即可完成Key的解析。

内容的提问来源于stack exchange，提问作者Kyley Jex