使用Mongo Sink Connector转换Kafka Header为Mongo字段时遇类型错误

问题背景 #

使用C# .NET 6开发服务，将事件发送至兼容Kafka协议的Azure EventHubs，Mongo Sink Connector消费功能正常。为让Connector读取事件Header，使用com.github.jcustenborder.kafka.connect.transform.common.HeaderToField$Value转换插件，配置如下：

{
  "transforms" : "headerToField",
  "transforms.headerToField.type" : "com.github.jcustenborder.kafka.connect.transform.common.HeaderToField$Value",
  "transforms.headerToField.header.mappings" : "applicationId:STRING"
}

运行时触发错误："MAP is not supported type"，但实际发送的是字符串类型Header。查看插件源码后确认，插件将Header识别为了Map类型，推测是Azure EventHubs基于AMQP协议发送Header时添加了额外字节导致。

原因分析 #

Azure EventHubs的Kafka协议实现底层依赖AMQP，发送Header时会自动为字符串添加AMQP类型标识的二进制封装，导致Kafka Connect接收到的Header并非单纯的字符串对象，而是被解析成了包含类型信息的Map结构，触发了插件中针对非预期类型的校验逻辑。

解决方法 #

1. 调整C#发送端的Header写入方式 #

使用Azure.Messaging.EventHubs发送事件时，跳过AMQP的自动类型封装，直接以原始字节形式写入Header：

using Azure.Messaging.EventHubs;
using System.Text;

var eventData = new EventData(Encoding.UTF8.GetBytes("your-event-payload"));
// 直接写入字符串的原始字节，避免AMQP类型标识
eventData.Properties["applicationId"] = Encoding.UTF8.GetBytes("your-application-id");

2. 改用原生Kafka客户端发送事件 #

放弃EventHubs专属客户端，使用Confluent.Kafka这类原生Kafka客户端发送事件，Header会以Kafka原生格式传递，不会被AMQP封装，插件可直接识别字符串类型：

using Confluent.Kafka;

var producerConfig = new ProducerConfig { BootstrapServers = "your-eventhubs-kafka-endpoint" };
using var producer = new ProducerBuilder<string, string>(producerConfig).Build();
var message = new Message<string, string>
{
    Key = "event-key",
    Value = "your-event-payload",
    Headers = new Headers { { "applicationId", Encoding.UTF8.GetBytes("your-application-id") } }
};
await producer.ProduceAsync("your-topic", message);

3. 自定义Kafka Connect转换插件 #

修改现有HeaderToField插件的类型校验逻辑，兼容AMQP封装的Header格式：

• 针对插件源码中类型判断的部分，增加对二进制Header的解析逻辑，提取出原始字符串后再执行映射；
• 或者自行实现一个轻量转换插件，先将AMQP封装的Header解析为字符串，再传递给原插件处理。

4. 使用Kafka Connect内置转换预处理Header #

通过内置转换先解析Header的二进制内容，再映射到字段：

{
  "transforms": "decodeHeader,headerToField",
  "transforms.decodeHeader.type": "org.apache.kafka.connect.transforms.InsertField$Value",
  "transforms.decodeHeader.static.field": "applicationId",
  "transforms.decodeHeader.static.value": "${file:bytesToString(${headers.applicationId})}",
  "transforms.headerToField.type": "com.github.jcustenborder.kafka.connect.transform.common.HeaderToField$Value",
  "transforms.headerToField.header.mappings": "applicationId:STRING"
}

（注：部分Kafka Connect环境可能需要扩展内置函数，或自定义简单转换来完成二进制到字符串的解码）

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