private RecordAppendResult tryAppend(long timestamp, byte[] key, byte[] value, Header[] headers,
                                         Callback callback, Deque<ProducerBatch> deque, long nowMs) {
        if (closed)
            throw new KafkaException("Producer closed while send in progress");
        ProducerBatch last = deque.peekLast();
        if (last != null) {
            int initialBytes = last.estimatedSizeInBytes();
            FutureRecordMetadata future = last.tryAppend(timestamp, key, value, headers, callback, nowMs);
            if (future == null) {
                last.closeForRecordAppends();
            } else {
                int appendedBytes = last.estimatedSizeInBytes() - initialBytes;
                return new RecordAppendResult(future, deque.size() > 1 || last.isFull(), false, false, appendedBytes);
            }
        }
        return null;
    }
Try to append to a ProducerBatch.

 If it is full, we return null and a new batch is created. We also close the batch for record appends to free up
 resources like compression buffers. The batch will be fully closed (ie. the record batch headers will be written
 and memory records built) in one of the following cases (whichever comes first): right before send,
 if it is expired, or when the producer is closed.
clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java