public abstract long getSizeOfQueuedBuffersUnsafe();
@Test
    void testSizeOfQueuedBuffers() throws Exception {
        // given: Configured pipelined result with 2 subpartitions.
        PipelinedResultPartition resultPartition =
                (PipelinedResultPartition) createResultPartition(ResultPartitionType.PIPELINED);

        ResultSubpartition[] subpartitions = resultPartition.subpartitions;
        assertThat(subpartitions).hasSize(2);

        PipelinedSubpartition subpartition0 = (PipelinedSubpartition) subpartitions[0];
        PipelinedSubpartition subpartition1 = (PipelinedSubpartition) subpartitions[1];

        // and: Set the buffers size.
        subpartition0.bufferSize(10);
        subpartition1.bufferSize(10);

        // when: Emit different records into different subpartitions.
        // Emit the recovered state.
        try (BufferBuilder bufferBuilder = getFinishedBufferBuilder(resultPartition, 6);
                BufferConsumer bufferConsumer = bufferBuilder.createBufferConsumerFromBeginning()) {
            resultPartition.addRecovered(0, bufferConsumer.copy());
            assertThat(resultPartition.getSizeOfQueuedBuffersUnsafe()).isEqualTo(6);

            resultPartition.addRecovered(1, bufferConsumer.copy());
            assertThat(resultPartition.getSizeOfQueuedBuffersUnsafe()).isEqualTo(12);
        }

        // Emit the EndOfChannelStateEvent for all channel
        resultPartition.finishReadRecoveredState(true);
        assertThat(resultPartition.getSizeOfQueuedBuffersUnsafe()).isEqualTo(20);

        // Emit the record less than buffer size.
        resultPartition.emitRecord(ByteBuffer.allocate(3), 0);
        assertThat(resultPartition.getSizeOfQueuedBuffersUnsafe()).isEqualTo(23);

        resultPartition.emitRecord(ByteBuffer.allocate(3), 1);
        assertThat(resultPartition.getSizeOfQueuedBuffersUnsafe()).isEqualTo(26);

        // Emit the record the equal to buffer size.
        resultPartition.emitRecord(ByteBuffer.allocate(10), 0);
        assertThat(resultPartition.getSizeOfQueuedBuffersUnsafe()).isEqualTo(36);

        resultPartition.emitRecord(ByteBuffer.allocate(10), 1);
        assertThat(resultPartition.getSizeOfQueuedBuffersUnsafe()).isEqualTo(46);

        // Broadcast event.
        resultPartition.broadcastEvent(EndOfPartitionEvent.INSTANCE, false);
        assertThat(resultPartition.getSizeOfQueuedBuffersUnsafe()).isEqualTo(54);

        // Emit one more record to the one subpartition.
        resultPartition.emitRecord(ByteBuffer.allocate(5), 0);
        assertThat(resultPartition.getSizeOfQueuedBuffersUnsafe()).isEqualTo(59);

        // Broadcast record.
        resultPartition.broadcastRecord(ByteBuffer.allocate(7));
        assertThat(resultPartition.getSizeOfQueuedBuffersUnsafe()).isEqualTo(73);

        // Poll the recovered state.
        assertThat(subpartition0.pollBuffer().buffer().getSize()).isEqualTo(6);
        assertThat(resultPartition.getSizeOfQueuedBuffersUnsafe()).isEqualTo(67);

        assertThat(subpartition1.pollBuffer().buffer().getSize()).isEqualTo(6);
        assertThat(resultPartition.getSizeOfQueuedBuffersUnsafe()).isEqualTo(61);

        // Poll the EndOfChannelStateEvent and resume the consumption.
        assertThat(subpartition0.pollBuffer().buffer().getSize()).isEqualTo(4);
        assertThat(resultPartition.getSizeOfQueuedBuffersUnsafe()).isEqualTo(57);

        assertThat(subpartition1.pollBuffer().buffer().getSize()).isEqualTo(4);
        assertThat(resultPartition.getSizeOfQueuedBuffersUnsafe()).isEqualTo(53);

        subpartition0.resumeConsumption();
        subpartition1.resumeConsumption();

        // when: Poll finished buffers.
        assertThat(subpartition0.pollBuffer().buffer().getSize()).isEqualTo(10);
        assertThat(resultPartition.getSizeOfQueuedBuffersUnsafe()).isEqualTo(43);

        assertThat(subpartition1.pollBuffer().buffer().getSize()).isEqualTo(10);
        assertThat(resultPartition.getSizeOfQueuedBuffersUnsafe()).isEqualTo(33);

        // Poll records which were unfinished because of broadcasting event.
        assertThat(subpartition0.pollBuffer().buffer().getSize()).isEqualTo(3);
        assertThat(resultPartition.getSizeOfQueuedBuffersUnsafe()).isEqualTo(30);

        assertThat(subpartition1.pollBuffer().buffer().getSize()).isEqualTo(3);
        assertThat(resultPartition.getSizeOfQueuedBuffersUnsafe()).isEqualTo(27);

        // Poll the event.
        assertThat(subpartition0.pollBuffer().buffer().getSize()).isEqualTo(4);
        assertThat(resultPartition.getSizeOfQueuedBuffersUnsafe()).isEqualTo(23);

        assertThat(subpartition1.pollBuffer().buffer().getSize()).isEqualTo(4);
        assertThat(resultPartition.getSizeOfQueuedBuffersUnsafe()).isEqualTo(19);

        // Poll the unfinished buffer.
        assertThat(subpartition0.pollBuffer().buffer().getSize()).isEqualTo(5);
        assertThat(resultPartition.getSizeOfQueuedBuffersUnsafe()).isEqualTo(14);

        // Poll broadcasted record.
        assertThat(subpartition0.pollBuffer().buffer().getSize()).isEqualTo(7);
        assertThat(resultPartition.getSizeOfQueuedBuffersUnsafe()).isEqualTo(7);

        assertThat(subpartition1.pollBuffer().buffer().getSize()).isEqualTo(7);
        assertThat(resultPartition.getSizeOfQueuedBuffersUnsafe()).isEqualTo(0);
    }