Kafka 1.1.0：如何获取消息的压缩后大小？

嘿，针对你遇到的这个压缩后消息大小确认的问题，我有几个实用的方法可以帮你搞定：

如果你想在发送前就知道大概的压缩后大小，可以在代码里手动对序列化后的消息做压缩，直接获取字节长度。虽然Kafka实际是批量压缩，单条压缩的效果和批量会略有差异，但用来快速估算完全够用。举个GZIP压缩的例子：

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.zip.GZIPOutputStream;

public class CompressionSizeEstimator {
    public static int getCompressedSize(byte[] uncompressedData) throws IOException {
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        try (GZIPOutputStream gzipOut = new GZIPOutputStream(baos)) {
            gzipOut.write(uncompressedData);
        }
        return baos.toByteArray().length;
    }
}

// 发送前调用
byte[] serializedMsg = yourMessageSerializer.serialize("your-topic", yourMessage);
int compressedSize = CompressionSizeEstimator.getCompressedSize(serializedMsg);
System.out.println("预估压缩后大小：" + compressedSize + "字节");

Kafka生产者的内部日志里会记录发送的批量数据大小，这个大小就是压缩后的实际大小（包含少量Kafka头部信息，几乎可以忽略）。你只需要把日志级别调到DEBUG：

• 如果用log4j，在配置里添加：

log4j.logger.org.apache.kafka.clients.producer.internals.Sender=DEBUG

• 如果用logback，添加：

<logger name="org.apache.kafka.clients.producer.internals.Sender" level="DEBUG"/>

之后你就能在日志里看到类似这样的条目：

Sending batch of 1 records with size 89234 bytes to topic your-topic partition 0

这里的89234就是压缩后的批量总大小，如果你是单条发送，这个数值就是单条压缩后的大小。

如果是测试阶段，你可以发送一条测试消息，然后用消费者读取并获取它的字节长度。这种方法能拿到最准确的实际存储大小，适合最终验证：

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class ActualMessageSizeChecker {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "your-broker:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "size-validation-group");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");

        try (Consumer<String, byte[]> consumer = new KafkaConsumer<>(props)) {
            TopicPartition tp = new TopicPartition("your-topic", 0);
            consumer.assign(Collections.singleton(tp));
            consumer.seekToBeginning(Collections.singleton(tp));
            
            ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofSeconds(3));
            records.forEach(record -> {
                int actualSize = record.value().length;
                System.out.println("实际压缩后消息大小：" + actualSize + "字节");
            });
        }
    }
}

因为Kafka是批量压缩，所以批量越大压缩效率通常越高。如果你的生产环境是批量发送消息，建议用日志里的批量大小作为参考，比单条估算更准确。另外，Kafka Connect的话，你可以在Connect的worker配置里开启相同的DEBUG日志，或者自定义一个Converter来记录压缩后的消息大小。

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