如何为Avro格式的Kafka消息添加消息头？

方案1：在Avro Schema中内嵌消息头字段 #

如果希望消息头和业务数据统一用Avro序列化，最直接的方式是修改Avro Schema，新增消息头相关字段来标识事件类型、版本等元信息，以此区分同一Kafka主题下的不同事件。

步骤1：修改Avro Schema文件 #

将原有事件结构封装为包含header（消息头）和payload（业务数据）的复合结构，示例如下：

{
  "type": "record",
  "name": "EnvelopedAvroMessage",
  "namespace": "com.example.avro",
  "fields": [
    {
      "name": "header",
      "type": {
        "type": "record",
        "name": "MessageHeader",
        "fields": [
          {"name": "eventType", "type": "string"}, // 事件类型标识，如"UserCreated"、"OrderUpdated"
          {"name": "version", "type": "string", "default": "1.0"}, // 消息版本
          {"name": "timestamp", "type": "long"} // 消息生成时间戳
        ]
      }
    },
    {
      "name": "payload",
      "type": [
        "null",
        {"type": "record", "name": "UserCreated", "fields": [/* 原用户事件字段定义 */]},
        {"type": "record", "name": "OrderUpdated", "fields": [/* 原订单事件字段定义 */]}
      ]
    }
  ]
}

通过header.eventType可以明确当前消息对应的事件类型，单元测试时可据此区分不同消息做针对性验证。

步骤2：调整Scala代码生成带消息头的Avro消息 #

修改原有方法，支持读取业务数据并手动注入消息头（或直接读取已包含头的Avro文件）：

import org.apache.avro.Schema
import org.apache.avro.generic.{GenericData, GenericDatumReader, GenericDatumWriter, GenericRecord}
import org.apache.avro.io.{BinaryEncoder, EncoderFactory}
import org.apache.avro.file.DataFileReader
import java.io.{ByteArrayOutputStream, File}
import scala.io.Source

def AvroKafkaMessageWithHeader(schemaPath: String, dataPath: String, eventType: String): Array[Byte] = {
  // 读取复合结构的Avro Schema
  val schemaStr = Source.fromFile(schemaPath).mkString
  val schemaObj = new Schema.Parser().parse(schemaStr)
  
  // 读取原始业务数据作为payload
  val payloadSchema = schemaObj.getField("payload").schema().getTypes.get(1)
  val payloadReader = new GenericDatumReader[GenericRecord](payloadSchema)
  val dataFileReader = new DataFileReader[GenericRecord](new File(dataPath), payloadReader)
  val payloadDatum = dataFileReader.next()
  
  // 构建消息头
  val headerSchema = schemaObj.getField("header").schema()
  val headerRecord = new GenericData.Record(headerSchema)
  headerRecord.put("eventType", eventType)
  headerRecord.put("version", "1.0")
  headerRecord.put("timestamp", System.currentTimeMillis())
  
  // 组装完整的包裹消息
  val envelopedRecord = new GenericData.Record(schemaObj)
  envelopedRecord.put("header", headerRecord)
  envelopedRecord.put("payload", payloadDatum)
  
  // 序列化完整消息
  val writer = new GenericDatumWriter[GenericRecord](schemaObj)
  val out = new ByteArrayOutputStream()
  val encoder: BinaryEncoder = EncoderFactory.get().binaryEncoder(out, null)
  writer.write(envelopedRecord, encoder)
  encoder.flush()
  out.close()
  
  out.toByteArray()
}

如果测试用的Avro文件已经包含完整的EnvelopedAvroMessage结构（自带header），可直接读取整个记录，无需手动构建头信息。

方案2：使用Kafka原生消息头（与Avro payload分离） #

若不想修改Avro Schema，可利用Kafka的RecordHeader机制添加元数据，消息头作为Kafka层面的属性与Avro业务数据分离。

步骤1：生成带Kafka头的ProducerRecord #

在发送消息时，给ProducerRecord添加自定义头：

import org.apache.kafka.clients.producer.ProducerRecord
import java.nio.charset.StandardCharsets

// 用原有方法生成Avro业务数据
val avroPayload = AvroKafkaMessage("path/to/schema.avsc", "path/to/data.avro")

// 创建带消息头的Kafka记录
val kafkaTopic = "your_topic_name"
val producerRecord = new ProducerRecord[String, Array[Byte]](kafkaTopic, avroPayload)
// 添加事件类型头
producerRecord.headers().add("event-type", "UserCreated".getBytes(StandardCharsets.UTF_8))
// 添加版本头
producerRecord.headers().add("version", "1.0".getBytes(StandardCharsets.UTF_8))

步骤2：单元测试中读取Kafka消息头 #

在单元测试（如ScalaTest结合嵌入式Kafka）中，消费消息后可读取头信息做验证：

import org.apache.kafka.clients.consumer.ConsumerRecord
import java.nio.charset.StandardCharsets

// 从测试Kafka集群获取消费记录
val consumerRecord: ConsumerRecord[String, Array[Byte]] = /* 消费逻辑获取 */
val eventType = consumerRecord.headers().lastHeader("event-type").value().toString(StandardCharsets.UTF_8)
val version = consumerRecord.headers().lastHeader("version").value().toString(StandardCharsets.UTF_8)

// 根据事件类型验证对应Avro数据的正确性
eventType match {
  case "UserCreated" => /* 验证用户事件结构 */
  case "OrderUpdated" => /* 验证订单事件结构 */
}

单元测试注意事项 #

• 方案1适合需要将元数据与业务数据统一序列化的场景，测试时可直接验证Avro结构的完整性，以及基于头字段的路由逻辑。
• 方案2适合无需修改Avro Schema的场景，测试时需验证Kafka消息头的传递正确性，以及基于头信息的事件区分逻辑。

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