如何通过Kafka Connect生成INT32类型的消息Key？

要让Kafka Connect把数据库ID字段序列化为4字节二进制整数，匹配Kafka Streams的Serdes.Integer()反序列化要求，你需要使用Kafka Connect内置的IntegerConverter，而非你尝试过的其他转换器。

问题根源分析 #

你之前尝试的转换器均无法生成符合IntegerDeserializer要求的二进制格式：

• StringConverter：把整数转为字符串字节（如123变成"123"的3字节数据），长度不符合IntegerDeserializer要求的4字节。
• JsonConverter：即使指定类型，仍会将整数序列化为JSON字符串格式，并非原始二进制整数。
• ByteArrayConverter：仅支持字节数组类型数据，与ExtractField输出的INT32 schema不兼容。
• AvroConverter：Avro序列化的整数带有自身编码格式，并非原始4字节二进制，且单独注册int schema会报错。

正确配置方案 #

修改Kafka Connect配置，将Key转换器指定为IntegerConverter，同时保留现有的transform规则：

# 保留抽取ID作为Key的transform配置
transforms.valueToKey.type: org.apache.kafka.connect.transforms.ValueToKey
transforms.valueToKey.fields: ID
transforms.extractId.type: org.apache.kafka.connect.transforms.ExtractField$Key
transforms.extractId.field: ID

# 使用IntegerConverter将Key序列化为二进制整数
key.converter: org.apache.kafka.connect.storage.IntegerConverter

旧版本Kafka Connect兼容方案 #

如果你的Kafka Connect版本较旧，没有IntegerConverter，可以使用PrimitiveConverter并指定类型：

key.converter: org.apache.kafka.connect.converters.PrimitiveConverter
key.converter.type: int

验证 #

配置生效后，Kafka Connect会将ID字段序列化为标准4字节二进制整数，此时Kafka Streams应用使用Serdes.Integer()即可正常反序列化Key，不会再出现Size of data received by IntegerDeserializer is not 4的错误。

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