Kafka 消费者 OutOfMemoryError: Java heap space —— 高压缩比消息「解压膨胀」撑爆小堆，fetch 限制只算压缩后字节

日期：2026-09-11 ｜题号：第 48 题｜分支：2026-09-11-1600 技术栈：Apache Kafka 3.9.0（KRaft 单节点）、Java 21（Temurin 21.0.12）、kafka-clients 3.9.0、Docker Compose、zstd / lz4 / gzip 压缩、JVM 堆内存关键词：Kafka 消费者 OOM、java.lang.OutOfMemoryError: Java heap space、DefaultRecord.readFrom、DefaultRecordBatch$2.doReadRecord、max.partition.fetch.bytes 无效、fetch.max.bytes 无效、max.poll.records、Kafka 压缩消息解压内存、高压缩比消费者内存爆炸、Kafka 调大堆、HeapByteBuffer allocate

一、问题背景

一套日志/事件采集链路，Kafka 磁盘占用很省，压缩效果惊人：

topic=logs-cmp-zstd   磁盘 147,937 字节 / 原始 5,850,911 字节   压缩比 39x

一个"压得很好的 topic"，单条 5MB 的大消息写进去在磁盘上只占 322 字节。

但新上线的消费者服务（容器内存吃紧，堆只给了 24MB）反复被 OOM 杀掉，日志里只有一行：

java.lang.OutOfMemoryError: Java heap space

堆栈不在业务代码，而在 Kafka 客户端内部：

at java.base/java.nio.HeapByteBuffer.<init>(HeapByteBuffer.java:71)
at java.base/java.nio.ByteBuffer.allocate(ByteBuffer.java:391)
at org.apache.kafka.common.record.DefaultRecord.readFrom(DefaultRecord.java:280)
at org.apache.kafka.common.record.DefaultRecordBatch$2.doReadRecord(DefaultRecordBatch.java:291)

最迷惑的地方：按"常识"去调 fetch 参数，完全没用。

max.partition.fetch.bytes = 1048576   → 还是 OOM
max.partition.fetch.bytes = 52428800  → 还是 OOM
max.partition.fetch.bytes = 104857600 → 还是 OOM
max.partition.fetch.bytes = 65536     → 还是 OOM（调小也 OOM，更反直觉）

fetch 明明限制在 1MB，为什么还能把 24MB 的堆撑爆？

二、复现步骤

2.1 一键环境（真实可跑）

仓库：scenario/2026-09-11-1600/，Docker Compose 三容器（broker / prepare / client）。

git fetch origin && git checkout 2026-09-11-1600
cd scenario/2026-09-11-1600
docker compose up -d --build                     # 约 1~2 分钟
docker compose exec client sh /scripts/run_all.sh

启动后预期看到：

 Container kafka-deflate-broker    Healthy
 Container kafka-deflate-prepare   Exited (0)
 Container kafka-deflate-client    Started

2.2 复现输出

对照组：未压缩 topic，24m 小堆读得动

[consumer] maxHeap=24MB topic=big-none-... 覆盖参数=(默认)
  poll 返回 1 条 / 解压后 5.0 MB（累计 1 条 / 5.0 MB）
  poll 返回 1 条 / 解压后 5.0 MB（累计 2 条 / 10.0 MB）
  poll 返回 1 条 / 解压后 5.0 MB（累计 3 条 / 15.0 MB）
DONE records=3 decompressed=15.0MB elapsed=4772ms

实验组：压缩 topic，必现 OOM

[consumer] maxHeap=24MB topic=big-zstd-... 覆盖参数=(默认)
Exception in thread "main" java.lang.OutOfMemoryError: Java heap space
    at java.base/java.nio.HeapByteBuffer.<init>(HeapByteBuffer.java:71)
    at org.apache.kafka.common.record.DefaultRecord.readFrom(DefaultRecord.java:280)

gzip / lz4 / zstd 三种 codec 全部 OOM；只有 none 正常。

2.3 复现矩阵

codec	消费者堆	结果
none	24m	✅ 正常读完 15MB
gzip	24m	💥 OutOfMemoryError
lz4	24m	💥 OutOfMemoryError
zstd	24m	💥 OutOfMemoryError
zstd	48m	✅ 实测下限
zstd	256m	✅ 正常

2.4 手工复现（进容器）

docker compose exec client bash
export CP="/app:$(ls /opt/kafka-libs/*.jar | tr '\n' ':')"

java -Xmx24m -cp "$CP" SmallHeapConsumer big-none-<ts>                 # ✅
java -Xmx24m -cp "$CP" SmallHeapConsumer big-zstd-<ts>                 # 💥
java -Xmx24m -cp "$CP" SmallHeapConsumer big-zstd-<ts> max.partition.fetch.bytes=52428800   # 💥 无效
java -Xmx24m -cp "$CP" SmallHeapConsumer big-zstd-<ts> max.poll.records=1                   # ✅ 有效

三、根本原因

3.1 一句话根因

max.partition.fetch.bytes / fetch.max.bytes 统计的是 broker 返回的「压缩后」字节。数据压缩比越高，它越"看起来正常"，而客户端拿到后会先整体解压，解压后的堆占用可达压缩前的上万倍。

3.2 数据链路（每一步都实测过）

用 kafka-dump-log.sh 直接看 record batch 头，压缩类型真实落在磁盘上：

docker compose exec broker bash -c '/opt/kafka/bin/kafka-dump-log.sh \
  --files /tmp/kafka-logs/big-zstd-<ts>-0/00000000000000000000.log 2>/dev/null \
  | grep -oE "size: [0-9]+ magic: 2 compresscodec: [a-z]+"'

size: 322 magic: 2 compresscodec: zstd
size: 322 magic: 2 compresscodec: zstd
size: 322 magic: 2 compresscodec: zstd

每条 batch：压缩后 322 字节 → 解压后 5,243,014 字节。压缩比 = 16,282 倍。

3.3 铁证：把 fetch 上限卡到字节级别

for v in 321 322 644 966 967; do
  java -Xmx512m -cp "$CP" SmallHeapConsumer big-zstd-<ts> "max.partition.fetch.bytes=$v"
done

实测每次 poll 返回的记录数：

max.partition.fetch.bytes=321  -> poll 返回 1 条 poll 返回 1 条 poll 返回 1 条
max.partition.fetch.bytes=322  -> poll 返回 1 条 poll 返回 1 条 poll 返回 1 条
max.partition.fetch.bytes=644  -> poll 返回 2 条 poll 返回 1 条
max.partition.fetch.bytes=966  -> poll 返回 3 条
max.partition.fetch.bytes=967  -> poll 返回 3 条

每次 poll 的记录数 = floor(fetch上限 ÷ 322)，边界严格由「压缩后」的 322 字节决定。

关键反直觉点：把上限设成 966（= 数据在磁盘上的全部字节数），客户端确实"只拿到 966 字节"，但解压出来仍是 15MB，24m 堆照样爆。

3.4 代码路径（javap 确认字节码）

javap -p -c -cp /opt/kafka-libs/kafka-clients-3.9.0.jar \
  org.apache.kafka.common.record.DefaultRecordBatch | grep -E "recordInputStream|wrapForInput"

public java.io.InputStream recordInputStream(BufferSupplier);
  19: invokestatic  // Compression.of:(...)Compression$Builder
  33: invokeinterface // Compression.wrapForInput:(...)InputStream
  28: invokespecial // DefaultRecordBatch$2."<init>":(...)

流程：recordInputStream() 先用 Compression.wrapForInput 把 322 字节解压成流，再逐条 DefaultRecord.readFrom 分配 ByteBuffer（走堆）→ OOM 就发生在这里。

同时，客户端并不会在解压前检查解压后大小是否超过任何 fetch 配置 —— 它没有这个信息（batch 头里不存解压后大小）。

四、解决办法

4.1 修复 A：`max.poll.records` 限制单次 poll 的解压量（首选，不改堆）

java -Xmx24m -cp "$CP" SmallHeapConsumer big-zstd-<ts> max.poll.records=1

  poll 返回 1 条 / 解压后 5.0 MB（累计 1 条 / 5.0 MB）
  poll 返回 1 条 / 解压后 5.0 MB（累计 2 条 / 10.0 MB）
  poll 返回 1 条 / 解压后 5.0 MB（累计 3 条 / 15.0 MB）
DONE records=3 decompressed=15.0MB elapsed=4830ms        ✅

堆保持 24m 不变，问题解决。门槛就在 1~2 条之间：

-Xmx24m + max.poll.records=1 -> ✅ DONE
-Xmx24m + max.poll.records=2 -> 💥 OutOfMemoryError
-Xmx24m + max.poll.records=3 -> 💥 OutOfMemoryError

4.2 修复 B：按「解压后字节数」重新评估堆

java -Xmx256m -cp "$CP" SmallHeapConsumer big-zstd-<ts>      # ✅

堆下限实测（默认参数，一次 poll 取完 3 条 = 15MB 解压后）：

-Xmx16m -> OOM
-Xmx24m -> OOM
-Xmx32m -> OOM
-Xmx48m -> ✅ DONE    ← 实测下限
-Xmx64m -> ✅

经验公式：堆需求 ≈ 解压后字节数 × 2~3。 15MB 解压数据约需 45MB，所以 48m 压线通过。注意不要按"磁盘占用"或"fetch 上限"估堆 —— 按压缩后大小估会低估上万倍。

4.3 治疗法：生产端拆消息（根治）

把一条 5MB 拆成 10 条 0.5MB 再发：

java -Xmx512m -cp "$CP" ChunkProducer big-chunked-<ts> 10 5242880
java -Xmx24m  -cp "$CP" SmallHeapConsumer big-chunked-<ts>

  poll 返回 10 条 / 解压后 5.0 MB（累计 10 条 / 5.0 MB）
DONE records=10 decompressed=5.0MB elapsed=4924ms        ✅

24m 小堆一次 poll 拿 10 条也没事 —— 单条解压后只有 0.5MB，总量可控。

4.4 监控侧

把「解压后字节数」（decompressed bytes）纳入观测，而不是只看 fetch 配置；
对「压缩比异常高」的 topic/分区告警；
同时盯消费者堆使用率与 GC 频率（OOM 前常先有长时间 Full GC 抖动）。

4.5 结论对照表

做法	是否有效	说明
调大 `max.partition.fetch.bytes`	❌	只管压缩字节，与解压膨胀无关
调小 `max.partition.fetch.bytes`	❌	最小也要给一个完整 batch，仍会解压出全量
调大/调小 `fetch.max.bytes`	❌	同上
`max.poll.records=1`	✅	限制单次解压量，堆不变即通过
调大堆到 48m+	✅	按解压后字节 × 2~3 估算
生产端拆小消息	✅	治本

五、验证

# 1) 一键脚本全流程（含复现 + 无效尝试 + 两种修复）
docker compose exec client sh /scripts/run_all.sh 2>&1 | tee /tmp/run_all.log

# 2) 修复参考实现三模式
export CP="/app:$(ls /opt/kafka-libs/*.jar | tr '\n' ':')"
java -Xmx24m  -cp "$CP" SafeConsumer big-zstd-<ts> buggy       # 💥 复现
java -Xmx24m  -cp "$CP" SafeConsumer big-zstd-<ts> pollLimit   # ✅ 修复
java -Xmx512m -cp "$CP" SafeConsumer big-zstd-<ts> diag        # 逐条评估

# 3) 清理
docker compose down -v

三模式实测：

[safe-consumer] mode=buggy     maxHeap=24MB  -> Exception in thread "main" java.lang.OutOfMemoryError: Java heap space
[safe-consumer] mode=pollLimit maxHeap=24MB  -> DONE records=3 decompressed=15.0MB heapCap=24MB   ✅
[safe-consumer] mode=diag      maxHeap=512MB -> 单条解压后 5.00MB, 堆上限 512MB -> 正常处理

完整操作过程与踩坑记录见 scenario/2026-09-11-1600/SOLUTION.md。

六、经验与避坑

压缩数据的内存成本 ≠ 磁盘成本。 Kafka 磁盘很省不等于消费者内存很省。压缩比越高，两者差距越大 —— 本案单条差 16,282 倍。
fetch 类参数统计的都是「压缩后」字节。 max.partition.fetch.bytes、fetch.max.bytes 都是。要约束"单次 poll 的解压量"，只能靠 max.poll.records。
参数"看起来调了没用"时，先确认参数真的生效了。 Kafka 客户端对拼错的配置项静默忽略（不报错、不告警）。用 max.partition.fetch.bytes 这类准确名字。
估堆要按解压后字节 × 2~3。 还有解压缓冲、ByteBuffer、对象头开销。实测 15MB 解压数据需 48m，24m 连 2 条 10MB 都扛不住。
max.poll.records 不能突破单条消息的物理下限。 单条解压后就 5MB，堆给 16m 时 max.poll.records=1 照样 OOM。根治还得靠生产端控制单条体积。
kafka-console-producer.sh 请用 --compression-codec zstd。 写 --producer-property compression.type=zstd 会被静默忽略，得到 codec=none 的"假复现"。
验证压缩是否真落盘，用 kafka-dump-log.sh 看 compresscodec: 字段，别只看文件大小。
OOM 堆栈落在 DefaultRecord.readFrom / DefaultRecordBatch$RecordIterator 时，优先怀疑压缩/解压，而不是业务反序列化代码。

七、最小可复用命令清单

# 环境
cd scenario/2026-09-11-1600 && docker compose up -d --build
docker compose exec client sh /scripts/run_all.sh
docker compose exec broker bash /scripts/show_compression_ratio.sh

# 看压缩后 vs 解压后
docker compose exec broker bash -c '/opt/kafka/bin/kafka-dump-log.sh \
  --files /tmp/kafka-logs/big-zstd-<ts>-0/00000000000000000000.log | grep compresscodec'

# 复现 / 定位 / 修复
export CP="/app:$(ls /opt/kafka-libs/*.jar | tr '\n' ':')"
java -Xmx24m  -cp "$CP" SmallHeapConsumer big-zstd-<ts>                 # 💥 复现
java -Xmx512m -cp "$CP" SmallHeapConsumer big-zstd-<ts> max.partition.fetch.bytes=322   # 证明边界在压缩字节
java -Xmx24m  -cp "$CP" SmallHeapConsumer big-zstd-<ts> max.poll.records=1             # ✅ 修复 A
java -Xmx48m  -cp "$CP" SmallHeapConsumer big-zstd-<ts>                                # ✅ 修复 B
java -Xmx512m -cp "$CP" ChunkProducer big-chunked-<ts> 10 5242880                       # ✅ 治疗法
docker compose down -v