@Override
    public void writeRecords(BaseRecords records) {
        if (records instanceof MemoryRecords) {
            flushPendingBuffer();
            addBuffer(((MemoryRecords) records).buffer());
        } else if (records instanceof UnalignedMemoryRecords) {
            flushPendingBuffer();
            addBuffer(((UnalignedMemoryRecords) records).buffer());
        } else {
            flushPendingSend();
            addSend(records.toSend());
        }
    }
Write a record set. The underlying record data will be retained
in the result of {@link #build()}. See {@link BaseRecords#toSend()}.

@param records the records to write
clients/src/main/java/org/apache/kafka/common/protocol/SendBuilder.java