private void writeLargeRecord(
            ByteBuffer record, int targetSubpartition, DataType dataType, boolean isBroadcast)
            throws IOException {
        // a large record will be spilled to a separated data region
        fileWriter.startNewRegion(isBroadcast);

        List<BufferWithSubpartition> toWrite = new ArrayList<>();
        Queue<MemorySegment> segments = new ArrayDeque<>(freeSegments);

        while (record.hasRemaining()) {
            if (segments.isEmpty()) {
                fileWriter.writeBuffers(toWrite);
                toWrite.clear();
                segments = new ArrayDeque<>(freeSegments);
            }

            int toCopy = Math.min(record.remaining(), networkBufferSize);
            MemorySegment writeBuffer = checkNotNull(segments.poll());
            writeBuffer.put(0, record, toCopy);

            NetworkBuffer buffer = new NetworkBuffer(writeBuffer, (buf) -> {}, dataType, toCopy);
            BufferWithSubpartition bufferWithSubpartition =
                    new BufferWithSubpartition(buffer, targetSubpartition);
            updateStatistics(bufferWithSubpartition, isBroadcast);
            toWrite.add(compressBufferIfPossible(bufferWithSubpartition));
        }

        fileWriter.writeBuffers(toWrite);
        releaseFreeBuffers();
    }
@TestTemplate
    void testWriteLargeRecord() throws Exception {
        int numBuffers = useHashDataBuffer ? 100 : 15;
        BufferPool bufferPool = globalPool.createBufferPool(numBuffers, numBuffers);
        SortMergeResultPartition partition = createSortMergedPartition(10, bufferPool);

        ByteBuffer recordWritten = generateRandomData(bufferSize * numBuffers, new Random());
        partition.emitRecord(recordWritten, 0);
        assertThat(bufferPool.bestEffortGetNumOfUsedBuffers())
                .isEqualTo(useHashDataBuffer ? numBuffers : 0);

        partition.finish();
        partition.close();

        ResultSubpartitionView view =
                partition.createSubpartitionView(new ResultSubpartitionIndexSet(0), listener);
        ByteBuffer recordRead = ByteBuffer.allocate(bufferSize * numBuffers);
        readData(
                new ResultSubpartitionView[] {view},
                bufferWithSubpartition -> {
                    Buffer buffer = bufferWithSubpartition.getBuffer();
                    int numBytes = buffer.readableBytes();

                    MemorySegment segment = MemorySegmentFactory.allocateUnpooledSegment(numBytes);
                    Buffer fullBuffer = ((CompositeBuffer) buffer).getFullBufferData(segment);
                    if (fullBuffer.isBuffer()) {
                        ByteBuffer byteBuffer =
                                ByteBuffer.allocate(fullBuffer.readableBytes())
                                        .put(fullBuffer.getNioBufferReadable());
                        recordRead.put((ByteBuffer) byteBuffer.flip());
                    }
                    fullBuffer.recycleBuffer();
                });
        recordWritten.rewind();
        recordRead.flip();
        assertThat(recordRead).isEqualTo(recordWritten);
    }