如何计算Kafka中AVRO消息的大小？

Kafka中使用AVRO格式的消息（尤其是通过Confluent Schema Registry序列化器处理的）并非纯AVRO序列化数据，而是带有固定结构的封装头。要计算其大小，需先明确消息组成结构，再针对性计算各部分长度之和。

一、核心结构与计算逻辑 #

标准Confluent AVRO消息由三部分组成：

• 1字节的魔法值（固定为0，用于标识Confluent格式的AVRO消息）
• 4字节的Schema ID（大端序，对应Schema Registry中该AVRO Schema的唯一ID）
• N字节的AVRO序列化后的实际业务数据

因此，Kafka AVRO消息总大小 = 1 + 4 + 纯AVRO数据序列化后的字节长度

二、代码实现示例 #

Java 实现（基于Confluent客户端与Apache Avro） #

方式1：手动序列化纯AVRO数据计算

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.DatumWriter;
import org.apache.avro.io.Encoder;
import org.apache.avro.io.EncoderFactory;
import org.apache.avro.specific.SpecificDatumWriter;
import java.io.ByteArrayOutputStream;

public class AvroSizeCalc {
    public static void main(String[] args) throws Exception {
        // 定义AVRO Schema
        String schemaJson = "{\"type\":\"record\",\"name\":\"User\",\"fields\":[{\"name\":\"id\",\"type\":\"int\"},{\"name\":\"name\",\"type\":\"string\"}]}";
        Schema schema = new Schema.Parser().parse(schemaJson);

        // 构造测试用AVRO记录
        GenericRecord userRecord = new GenericData.Record(schema);
        userRecord.put("id", 456);
        userRecord.put("name", "Bob");

        // 序列化纯AVRO数据并获取长度
        ByteArrayOutputStream outStream = new ByteArrayOutputStream();
        DatumWriter<GenericRecord> datumWriter = new SpecificDatumWriter<>(schema);
        Encoder encoder = EncoderFactory.get().binaryEncoder(outStream, null);
        datumWriter.write(userRecord, encoder);
        encoder.flush();
        int avroPayloadLen = outStream.toByteArray().length;

        // 计算Kafka AVRO消息总大小
        int totalSize = 1 + 4 + avroPayloadLen;
        System.out.println("Kafka AVRO消息总大小: " + totalSize + " bytes");
    }
}

方式2：直接使用Confluent序列化器获取完整消息长度

import io.confluent.kafka.serializers.KafkaAvroSerializer;
import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.clients.producer.ProducerConfig;
import java.util.Properties;

public class KafkaAvroSizeCalc {
    public static void main(String[] args) {
        // 配置序列化器参数
        Properties config = new Properties();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put("schema.registry.url", "http://localhost:8081");

        KafkaAvroSerializer serializer = new KafkaAvroSerializer();
        serializer.configure(config, false);

        // 假设已构造好GenericRecord类型的userRecord
        byte[] fullKafkaAvroBytes = serializer.serialize("test-topic", userRecord);
        int totalSize = fullKafkaAvroBytes.length;
        System.out.println("Kafka AVRO消息总大小: " + totalSize + " bytes");
    }
}

Python 实现（基于confluent-kafka与avro库） #

import avro.schema
from avro.io import DatumWriter, BinaryEncoder
import io
from confluent_kafka.avro import AvroProducer

# 定义AVRO Schema
schema_json = """
{
    "type": "record",
    "name": "User",
    "fields": [
        {"name": "id", "type": "int"},
        {"name": "name", "type": "string"}
    ]
}
"""
schema = avro.schema.parse(schema_json)
test_user = {"id": 789, "name": "Charlie"}

# 方式1：手动计算纯AVRO数据长度再加固定头
writer = DatumWriter(schema)
byte_buffer = io.BytesIO()
encoder = BinaryEncoder(byte_buffer)
writer.write(test_user, encoder)
avro_payload_len = len(byte_buffer.getvalue())
total_size = 1 + 4 + avro_payload_len
print(f"Kafka AVRO消息总大小（手动计算）: {total_size} bytes")

# 方式2：用AvroProducer序列化后直接取长度
producer_config = {
    "bootstrap.servers": "localhost:9092",
    "schema.registry.url": "http://localhost:8081"
}
producer = AvroProducer(producer_config)
full_avro_bytes = producer.serializer.encode_record_with_schema(
    "test-topic", schema, test_user
)
print(f"Kafka AVRO消息总大小（序列化后获取）: {len(full_avro_bytes)} bytes")

三、关键注意事项 #

• 如果使用自定义AVRO序列化器（非Confluent标准实现），需确认是否有额外封装头，若没有则总大小就是纯AVRO数据的序列化长度
• 计算结果需匹配Kafka集群的message.max.bytes配置，避免因消息过大导致发送失败
• Schema ID的字节数固定为4，魔法值固定为1，无需考虑序列化细节，直接累加即可

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