程序报错如下：
解决方案：
调整kafka对单条消息的大小限制：message.max.bytes从默认1M调整为10M，10M是cdh建议的最大值了，调整过大会影响kafka的吞吐量,同时需要调整replica.fetch.max.bytes的值需要大于等于message.max.bytes，该参数为broker同步的消息大小限制，如果小于message.max.bytes可能会出现broker接收消息但是主副本无法同步从而造成数据丢失。（我们用着的cdh的管理在界面可以调整整个集群的默认参数，如果是单个kafka集群可以后天调整或者用kafkamanager调整）
详细介绍可参考：https://www.cnblogs.com/qiumingcheng/p/5631309.html