private void writeLargeRecord(ByteBuffer record, int subpartitionId, Buffer.DataType dataType) {

        checkState(dataType != Buffer.DataType.EVENT_BUFFER);
        while (record.hasRemaining()) {
            int toCopy = Math.min(record.remaining(), bufferSizeBytes);
            MemorySegment writeBuffer = requestBuffer().getMemorySegment();
            writeBuffer.put(0, record, toCopy);

            int numRemainingConsecutiveBuffers =
                    (int) Math.ceil(((double) record.remaining()) / bufferSizeBytes);
            if (numRemainingConsecutiveBuffers == 0) {
                dataType = Buffer.DataType.DATA_BUFFER_WITH_CLEAR_END;
            }

            flushBuffer(
                    new BufferWithSubpartition(
                            new NetworkBuffer(
                                    writeBuffer, checkNotNull(bufferRecycler), dataType, toCopy),
                            subpartitionId),
                    numRemainingConsecutiveBuffers);
        }

        releaseFreeBuffers();
    }
@Test
    void testWriteLargeRecord() throws IOException {
        testWriteLargeRecord(true);
    }