其实我有点犹豫要不要写这个文章，因为笔者的kafka客户端版本并不是最新的，不知道新版本是否会有这个问题。
版本提示
kafka server版本
Scala 2.11 - kafka_2.11-2.0.0.tgz
kafka client版本
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<version>2.1.7.RELEASE</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>1.1.1</version>
</dependency>
问题现象
笔者在优化公司kafka producer性能时发现一个很奇怪的现象，当 1000tps 从外部输入消息给kafka producer的时候，即每秒1000个record，但实际上会发现kafka.producer:type=producer-metrics mbean的record-send-rate会达到14000到15000r/s，record-error-rate会达到1000r/s。因为配置了batch.size为5MB和linger.ms为5秒，所以消息是会缓冲一段时间发送出去的，但是不至于能长时间持续积攒出每秒1万多的记录，这样大的发送速度发下去，也没消息能积攒下来啊。
我发现kafka客户端的日志没有打印，于是配置了org.apache.kafka.clients logger, 日志级别设置为TRACE。
我在日志中发现了下面的报错，关键词是MESSAGE_TOO_LARGE。
2020-05-14 21:29:58.624 [kafka-producer-network-thread | producer-1] WARN (org.apache.kafka.clients.producer.internals.Sender:509) completeBatch - [Producer clientId=producer-1] Got error produce response in correlation id 226233 on topic-partition span-1, splitting and retrying (3 attempts left). Error: MESSAGE_TOO_LARGE
后续还试过batch.size设置为1MB，也是一样的现象，凭空多出很多消息的感觉。
首次确定报错含义
搜索这个报错，是kafka broker侧有个配置message.max.bytes，默认为1MB，如果发送的消息体积大于这个，就会报错。
《kafka权威指南》说这个是限制单条record消息体积大小的。
官方文档 也是这么说的
The request included a message larger than the max message size the server will accept.
这些都指向request里面某个record超过了kafka broker侧message.max.bytes配置。
当服务端返回MESSAGE_TOO_LARGE报错后，客户端将消息拆分成更小的批次进行重新发送，所以会凭空产生很多消息的错觉。没去探究拆分逻辑，是怎么从1000r/s拆到15000r/s的也是奇怪，最好合理设置参数避免这种场景。
确定kafka broker侧message.max.bytes配置大小
开始我想通过kafka自带的kafka-config.sh，却发现在我这个版本，这个脚本带的命令不会反馈所有配置，只会返回你动态修改过的那些。
kafka-manager界面点击topic的配置，也只会显示空
通过搜索上面这个问题，我发现了 Getting broker configuration via kafka-configs.sh
Yes this is a known behavior: https://issues.apache.org/jira/browse/KAFKA-7720
Currently this tool only shows configurations that have been dynamically overridden.
To see all broker configs, you have 2 options:
Use the AdminClient.describeConfigs() API
Tweak the ConfigCommand.scala tool. Just remove the filter on https://github.com/apache/kafka/blob/ad26914de65e2db244fbe36a2a5fd03de87dfb79/core/src/main/scala/kafka/admin/ConfigCommand.scala#L347
Update: This issue is being addressed by KIP-524 which is currently targetted for Kafka 2.5
后面我到server.log去grep
我在server.log里发现了这个配置的大小kafka实际生效的message.max.bytes值是1000012字节，0.95左右，小于1MB。
bash-4.4# cat server.log.*|grep message.max.bytes
message.max.bytes = 1000012
message.max.bytes = 1000012
message.max.bytes = 1000012
message.max.bytes = 1000012
message.max.bytes = 1000012
message.max.bytes = 1000012
topic层的配置没有进行覆盖，所以参数值确定为1000012字节
疑问
但实际上我这次压测采用的消息体积是固定的，kafka producer metrics的record-size-avg是2.7KB，record-size-max是9.6KB，远远小于1000012字节。
为什么这样还会被拒绝呢。
调整batch.size重新测试
我将batch.size调整到512KB, max.request.size调整到2MB,重新压测，这次日志里不再刷MESSAGE_TOO_LARGE报错了。
record-send-rate恢复到1000r/s每秒，record-error-rate为0.问题消除
这里我已经感觉message.max.bytes可能限制的不是单条消息的大小，而是限制的batch.size的大小。
去kafka源码探究message.max.bytes
https://github.com/apache/kafka/blob/847ff8f/core/src/main/scala/kafka/log/Log.scala#L1377
// Check if the message sizes are valid.
val batchSize = batch.sizeInBytes
if (batchSize > config.maxMessageSize) {
brokerTopicStats.topicStats(topicPartition.topic).bytesRejectedRate.mark(records.sizeInBytes)
brokerTopicStats.allTopicsStats.bytesRejectedRate.mark(records.sizeInBytes)
throw new RecordTooLargeException(s"The record batch size in the append to $topicPartition is $batchSize bytes " +
s"which exceeds the maximum configured value of ${config.maxMessageSize}.")
}
不知道我这段看的对不对，不过确实非常可能是
message.max.bytes是限制的batch.size的大小。
总结
batch.size最好比服务端实际生效的message.max.bytes配置小一些，否则在笔者所在的版本会遇到MESSAGE_TOO_LARGE报错，客户端进行重试发送造成很大的消息发送量。这个重试是可以避免发生的。
