Kafka发送消息时提示请求数据过大是怎么回事？

// ...
// 估算消息的字节大小int serializedSize = AbstractRecords.estimateSizeInBytesUpperBound(apiVersions.maxUsableProduceMagic(),                                                                   compressionType, serializedKey, serializedValue, headers);
// 确保消息大小不超过发送请求最大值 max.request.size，或者发送缓冲池发小 buffer.memoryensureValidRecordSize(serializedSize);long timestamp = record.timestamp() == null ? time.milliseconds() : record.timestamp();log.trace("Sending record {} with callback {} to topic {} partition {}", record, callback, record.topic(), partition);// producer callback will make sure to call both 'callback' and interceptor callbackCallback interceptCallback = this.interceptors == null ? callback : new InterceptorCallback<>(callback, this.interceptors, tp);
if (transactionManager != null && transactionManager.isTransactional())  transactionManager.maybeAddPartitionToTransaction(tp);
// 向 RecordAccumulator 追加消息RecordAccumulator.RecordAppendResult result = accumulator.append(tp, timestamp, serializedKey,                                                                 serializedValue, headers, interceptCallback, remainingWaitMs);
// 如果 batch 已经满了，那么就会唤醒sender线程发送批量消息if (result.batchIsFull || result.newBatchCreated) {  log.trace("Waking up the sender since topic {} partition {} is either full or getting a new batch", record.topic(), partition);  this.sender.wakeup();}return result.future;// ...

private void ensureValidRecordSize(int size) {  // 如果发送消息的大小比maxRequestSize大，就会抛异常  if (size > this.maxRequestSize)    throw new RecordTooLargeException("The message is " + size +                                      " bytes when serialized which is larger than the maximum request size you have configured with the " +                                      ProducerConfig.MAX_REQUEST_SIZE_CONFIG +                                      " configuration.");  if (size > this.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.");}