private void emit(ByteBuffer record, int targetSubpartition, Buffer.DataType dataType)
            throws IOException {
        checkInProduceState();
        checkNotNull(memoryDataManager).append(record, targetSubpartition, dataType);
    }
@Test
    void testEmit() throws Exception {
        int numBuffers = 100;
        int numSubpartitions = 10;
        int numRecords = 1000;
        Random random = new Random();

        BufferPool bufferPool = globalPool.createBufferPool(numBuffers, numBuffers);

        try (HsResultPartition partition = createHsResultPartition(numSubpartitions, bufferPool)) {
            Queue<Tuple2<ByteBuffer, Buffer.DataType>>[] dataWritten = new Queue[numSubpartitions];
            Queue<Buffer>[] buffersRead = new Queue[numSubpartitions];
            for (int i = 0; i < numSubpartitions; ++i) {
                dataWritten[i] = new ArrayDeque<>();
                buffersRead[i] = new ArrayDeque<>();
            }

            int[] numBytesWritten = new int[numSubpartitions];
            int[] numBytesRead = new int[numSubpartitions];
            Arrays.fill(numBytesWritten, 0);
            Arrays.fill(numBytesRead, 0);

            for (int i = 0; i < numRecords; ++i) {
                ByteBuffer record = generateRandomData(random.nextInt(2 * bufferSize) + 1, random);
                boolean isBroadCast = random.nextBoolean();

                if (isBroadCast) {
                    partition.broadcastRecord(record);
                    for (int subpartition = 0; subpartition < numSubpartitions; ++subpartition) {
                        recordDataWritten(
                                record,
                                dataWritten,
                                subpartition,
                                numBytesWritten,
                                Buffer.DataType.DATA_BUFFER);
                    }
                } else {
                    int subpartition = random.nextInt(numSubpartitions);
                    partition.emitRecord(record, subpartition);
                    recordDataWritten(
                            record,
                            dataWritten,
                            subpartition,
                            numBytesWritten,
                            Buffer.DataType.DATA_BUFFER);
                }
            }

            partition.finish();

            for (int subpartition = 0; subpartition < numSubpartitions; ++subpartition) {
                ByteBuffer record = EventSerializer.toSerializedEvent(EndOfPartitionEvent.INSTANCE);
                recordDataWritten(
                        record,
                        dataWritten,
                        subpartition,
                        numBytesWritten,
                        Buffer.DataType.END_OF_PARTITION);
            }

            Tuple2<ResultSubpartitionView, TestingBufferAvailabilityListener>[] viewAndListeners =
                    createSubpartitionViews(partition, numSubpartitions);
            readData(
                    viewAndListeners,
                    (buffer, subpartitionId) -> {
                        int numBytes = buffer.readableBytes();
                        numBytesRead[subpartitionId] += numBytes;

                        MemorySegment segment =
                                MemorySegmentFactory.allocateUnpooledSegment(numBytes);
                        segment.put(0, buffer.getNioBufferReadable(), numBytes);
                        buffersRead[subpartitionId].add(
                                new NetworkBuffer(
                                        segment, (buf) -> {}, buffer.getDataType(), numBytes));
                    });
            checkWriteReadResult(
                    numSubpartitions, numBytesWritten, numBytesRead, dataWritten, buffersRead);
        }
    }