《Java 消息中间件·可靠消息与消费治理》11.1 消息大小与批量参数

尺寸不是一个上限，是四道闸

问「Kafka 一条消息最大能有多大」，得到的答案通常是一个数字，比如 1 MB。这个答案既不完整，也会误导排障：真正决定消息能不能发出去的是四层参数——其中三道会拦截，一道只改变批次——每层计量的对象不同，报错的位置和措辞也不同。

四个数字口径不同，所以「把 message.max.bytes 调大」并不等于「客户端就能发大消息」。反过来，一条 2 MiB 的消息也可能因为压缩后远小于 1 MiB 而顺利通过 broker 上限。本篇用一组逐级放大的实验把这四层拆开。

本机实测环境（Temurin 25.0.5+8 / kafka-clients 4.3.0 / apache/kafka:4.3.0 单节点 KRaft / 2026-10-29）：宿主机 Java 客户端连 127.0.0.1:19892，容器内 CLI 连 127.0.0.1:9092。所有耗时都是单次运行，不作基准。

客户端的第一道闸：max.request.size

KafkaProducer.doSend 在把记录交给累加器之前，会先算一个序列化大小的上界，再和 max.request.size 比较。源码原文（KafkaProducer.java，4.3.0）：

private void ensureValidRecordSize(int size) {
    if (size > maxRequestSize)
        throw new RecordTooLargeException("The message is " + size +
                " bytes when serialized which is larger than " + maxRequestSize + ", which is the value of the " +
                ProducerConfig.MAX_REQUEST_SIZE_CONFIG + " configuration.");
    if (size > totalMemorySize)
        throw new RecordTooLargeException("The message is " + size +
                " bytes when serialized which is larger than the total memory buffer you have configured with the " +
                ProducerConfig.BUFFER_MEMORY_CONFIG + " configuration.");
}

估算值包含 key、headers 和记录头开销，所以比 value 本身大几十字节。本机实测：65,536 字节的 value 估成 65,630（+94），1,048,576 字节的 value 估成 1,048,673（+97），2,097,152 字节估成 2,097,249（+97）。

一个容易写错的细节是异常抛在哪。旧资料常说 send() 会直接抛 RecordTooLargeException，但在 4.3.0 的源码里，RecordTooLargeException 继承 ApiException，而 doSend 的 catch (ApiException e) 分支返回的是 new FutureFailure(e)：

} catch (ApiException e) {
    log.debug("Exception occurred during message send:", e);
    if (callback != null) {
        TopicPartition tp = appendCallbacks.topicPartition();
        RecordMetadata nullMetadata = new RecordMetadata(tp, -1, -1, RecordBatch.NO_TIMESTAMP, -1, -1);
        callback.onCompletion(nullMetadata, e);
    }
    this.errors.record();
    this.interceptors.onSendError(record, appendCallbacks.topicPartition(), e);
    if (transactionManager != null) {
        transactionManager.maybeTransitionToErrorState(e);
    }
    return new FutureFailure(e);

实测完全一致：send() 正常返回，失败直到 future.get() 才暴露。下面每一条输出里的 send() returned 都是这个意思。

java -cp "$CP:out" SizeProbe size.probe 1048576

valueBytes=1048576 overrides={}
send() returned, elapsedSendUs=215479
FUTURE_FAIL class=org.apache.kafka.common.errors.RecordTooLargeException
FUTURE_FAIL errors=MESSAGE_TOO_LARGE code=10
FUTURE_FAIL message=The message is 1048673 bytes when serialized which is larger than 1048576, which is the value of the max.request.size configuration.

保持 max.request.size 默认 1 MiB，逐级放大 value：

注意 1,048,000 能过、1,048,576 不能，分界不在「1 MB」这个整数上，而在 1,048,576 减去记录开销之后。业务侧若按「value 不超过 1 MB」做校验，仍会有一小段区间在客户端炸掉。

放宽客户端，才看见 broker 的闸

把客户端上限调到 8 MiB，同一条 1 MiB 的 value 就会走到 broker，然后被另一句话拒绝：

java -cp "$CP:out" SizeProbe size.probe 1048576 max.request.size=8388608

valueBytes=1048576 overrides={max.request.size=8388608}
send() returned, elapsedSendUs=593390
FUTURE_FAIL class=org.apache.kafka.common.errors.RecordTooLargeException
FUTURE_FAIL errors=MESSAGE_TOO_LARGE code=10
FUTURE_FAIL message=The request included a message larger than the max message size the server will accept.

两层的异常类名和错误码完全一样，都是 RecordTooLargeException / MESSAGE_TOO_LARGE（错误码 10，Errors.forException 按异常类型映射，不区分是谁抛的）。区分只能靠消息文本：客户端说「larger than … max.request.size」，broker 说「larger than the max message size the server will accept」。排障时先看这句话，再去查对应那一层的配置，不要只看异常类名。

broker 默认 message.max.bytes=1048588。实测边界（客户端上限 8 MiB）：

按 value 字节算，边界落在 1,048,500 与 1,048,540 之间；换算成 batch 字节，就落在默认的 message.max.bytes=1048588 附近。差值来自批次头、key 和记录头：broker 计量的是整个 RecordBatch，而 message.max.bytes 约束的正是这个 batch 大小。

另一个实测事实：broker 拒绝这次写入后，/opt/kafka/logs/server.log 里没有对应的 INFO 记录（grep -i "too large" 无输出）。服务端拒绝的证据只在客户端异常里。不要把「broker 日志没报错」当成「消息发出去了」。

topic 级上限：客户端根本不知道

topic 可以单独覆盖 max.message.bytes。建一个上限 512 KiB 的 topic：

kafka-configs.sh --bootstrap-server 127.0.0.1:9092 \
  --entity-type topics --entity-name size.topiclimit --describe

Dynamic configs for topic size.topiclimit are:
  max.message.bytes=524288 sensitive=false synonyms={DYNAMIC_TOPIC_CONFIG:max.message.bytes=524288, DEFAULT_CONFIG:message.max.bytes=1048588}

输出里的 DEFAULT_CONFIG:message.max.bytes=1048588 说明 topic 配置的默认值来自 broker 的 message.max.bytes，两级是继承关系。实测：400,000 字节成功，524,288 字节被 broker 拒绝，报的仍是那句通用的 The request included a message larger than the max message size the server will accept.。

这里有个结构性限制：producer 的元数据里没有 topic 的 max.message.bytes。客户端只能按自己的 max.request.size 先放行，到了 broker 才知道这个 topic 更严。所以多 topic 共用一套生产者配置时，大消息会在不同 topic 上表现不一致，而客户端侧无法提前判断。

broker 级上限也可以动态改：

kafka-configs.sh --bootstrap-server 127.0.0.1:9092 \
  --entity-type brokers --entity-name 1 \
  --alter --add-config message.max.bytes=4194304

改完后以每秒一次的新客户端探测 2 MiB 消息，本次观察到 1 秒内生效（改后约 1 秒的首次探测即成功，落到 offset 9，耗时 310 ms）。动态配置不承诺立即生效，客户端元数据刷新有窗口；生产中改完要留出观察时间，不能改完立刻断言失败消失。4 MiB 的 value 在 4 MiB 的上限下仍然失败，因为 batch 本身还要加上批次和记录头。

压缩改变了「字节」的含义

broker 的 message.max.bytes 计量的是压缩后的 batch 字节。这一条会让直觉完全失效。

保持 broker 默认 1 MiB，客户端上限 8 MiB，发一条 2 MiB、内容全是重复字符的 value，压缩算法用 lz4：

valueBytes=2097152 overrides={compression.type=lz4, max.request.size=8388608}
send() returned, elapsedSendUs=3906948
OK partition=0 offset=10 serializedKeySize=9 serializedValueSize=2097152 elapsedMs=3970
（3970 ms 包含首次加载 lz4 本地库的时间，不是单纯的压缩耗时）

2 MiB 的 value 通过了 1 MiB 的 broker 上限。换成同样 2 MiB 的随机字节，压缩几乎不缩小，broker 立刻拒绝：

valueBytes=2097152 overrides={compression.type=lz4, max.request.size=8388608}
FUTURE_FAIL class=org.apache.kafka.common.errors.RecordTooLargeException
FUTURE_FAIL errors=MESSAGE_TOO_LARGE code=10
FUTURE_FAIL message=The request included a message larger than the max message size the server will accept.

而客户端的 max.request.size 检查用的是压缩前的估算上界。同一份 2 MiB 重复内容，只要 max.request.size 保持默认 1 MiB，客户端就先拦下来：

FUTURE_FAIL message=The message is 2097249 bytes when serialized which is larger than 1048576, which is the value of the max.request.size configuration.

两层口径不同步：客户端按「压缩前的最坏情况」保守拦截，broker 按「压缩后的真实 batch」放行。结果就是调参时必须同时想清楚两件事——客户端上限决定「业务允许多大的 payload」，broker 上限决定「网络和磁盘能承受多大的 batch」。把两者设成同一个数字，通常不是最优解。

batch.size 不是消息大小上限

batch.size 容易被当成「一批最多这么大」，实际语义是「攒到这么大就提前发」。累加器给新记录分配缓冲区时取两者较大值（RecordAccumulator.append，4.3.0）：

int size = Math.max(this.batchSize, AbstractRecords.estimateSizeInBytesUpperBound(
        RecordBatch.CURRENT_MAGIC_VALUE, compression.type(), key, value, headers));

所以一条 900 KiB 的记录配 batch.size=1024 照样能发：

valueBytes=921600 overrides={batch.size=1024}
send() returned, elapsedSendUs=218687
OK partition=0 offset=8 serializedKeySize=8 serializedValueSize=921600 elapsedMs=270

batch.size 也救不了超过 max.request.size 的记录。把 batch.size 调到 1 MiB、max.request.size 压到 32 KiB，发 64 KiB：

valueBytes=65536 overrides={max.request.size=32768, batch.size=1048576}
FUTURE_FAIL message=The message is 65630 bytes when serialized which is larger than 32768, which is the value of the max.request.size configuration.

反过来，max.request.size 也不是「一批的上限」。同在 4.3.0，一批比请求上限大时，只要它是该分区唯一待发的批次，仍会单独成一个请求发出去。RecordAccumulator.drainBatchesForOneNode 的注释写得很直白：

if (size + first.estimatedSizeInBytes() > maxSize && !ready.isEmpty()) {
    // there is a rare case that a single batch size is larger than the request size due to
    // compression; in this case we will still eventually send this batch in a single request
    break;
}

实测：batch.size=65536、max.request.size=32768，同一个 key 连发 20 条 8,000 字节的记录，全部成功；生产者指标 batch-split-total=0，request-total 合计为 7。结论是 max.request.size 管的是单条记录和请求的拼装，不是批次硬上限。

三个参数的职责可以这样记：

消费端还有一道，而且它会放行

生产端的三道关都过了，消费端还有 fetch.max.bytes 和 max.partition.fetch.bytes。直觉是「把 fetch 上限调小就能挡住大消息」，实测相反：

fetch.max.bytes=1024 max.partition.fetch.bytes=1024
received key=k-921600 partition=0 offset=0 valueBytes=921600
received key=k-2048 partition=0 offset=1 valueBytes=2048
total=2

fetch 上限设为 1 KiB，921,600 字节的记录仍然被完整返回。原因是 Kafka 为了保证消费不停滞，第一个批次即使超过 fetch 上限也会被返回。这意味着大消息不会被消费端拒绝，只会原样进入消费者的堆内存。想控制大消息，只能在生产端和 broker 端设限，不能指望消费端兜底。

排障顺序与配置建议

遇到 RecordTooLargeException，按下面顺序定位，不要先改 broker：

1. 看异常消息。出现 max.request.size 就是客户端拦截，改生产者配置或拆分消息。
2. 出现 max message size the server will accept 就是 broker 拦截；再确认是 topic 覆盖还是 broker 默认：kafka-configs.sh --describe 两级都看。
3. 若客户端和 broker 都已放宽仍失败，检查压缩：同样的 payload，压缩后是否真的变小；随机数据或已压缩数据不会变小。
4. 检查是不是 DLT、重试主题等旁路 topic 沿用了更小的上限，导致主链路能写、死信写不进去。

配置上的最小自洽条件：

• broker message.max.bytes 不小于任何 topic 的 max.message.bytes，否则 topic 配置只是摆设。
• 主主题与它的 DLT、重试主题保持同一上限，避免「原消息能进、死信进不去」。
• 生产者 max.request.size 至少覆盖「单条最大 payload + 100 字节左右开销」；它不是越大越好，过大会让单次请求占用更多内存和网络时间。
• 超过 1 MB 的消息先问一句「它该不该走消息」，答案通常是不该。11.3 节给出 claim-check 的替代方案。

复现本篇的阶梯实验

实验代码只有一个 SizeProbe，参数是「topic + value 字节数 + 可选的 producer 覆盖项」，每次起一个独立 JVM，避免缓冲区状态串扰：

java -cp "$CP:out" SizeProbe size.probe 1048576
java -cp "$CP:out" SizeProbe size.probe 1048576 max.request.size=8388608
java -cp "$CP:out" SizeProbe size.probe 1048576 max.request.size=8388608 compression.type=lz4

三段分别是：客户端拦截、broker 拦截、压缩后放行。topic 级上限用 kafka-configs.sh --entity-type topics --alter --add-config max.message.bytes=... 建，broker 级用 --entity-type brokers --entity-name 1。每换一档都要重新起 JVM，否则上一轮的 max.request.size 会留在客户端里，观察到的「层级」就不可信。

回到 libraryd

library.loan.created 的 note 字段是唯一可能长到几百 KB 的字段。按本篇的闸门口径，团队定的规则是：单条事件序列化后不超过 64 KiB，超过就把 note 放到对象存储，消息里只留指针（11.3 节展开）。生产者的 max.request.size 保持默认 1 MiB，broker 与 topic 的 max.message.bytes 统一为 1 MiB，DLT 与主主题一致。

验收时不要只测「正常消息能发」，要按本篇的阶梯各测一档：刚好低于客户端上限、刚好高于客户端上限、高于 broker 上限、以及一条压缩后能过但压缩前超限的消息。只有第四档能证明团队理解了两层口径的差异。

阅读导航：上一节 10.3 重分区与历史数据 ｜ 下一节 11.2 压缩算法选择