@Nullable
    @Override
    public ResultSubpartition.BufferAndBacklog getNextBuffer() throws IOException {
        synchronized (lock) {
            cacheBuffer();
            if (cachedBuffers.isEmpty()) {
                return null;
            }

            ResultSubpartition.BufferAndBacklog buffer = cachedBuffers.poll().f0;

            return new ResultSubpartition.BufferAndBacklog(
                    buffer.buffer(),
                    cachedBuffers.size(),
                    cachedBuffers.isEmpty()
                            ? Buffer.DataType.NONE
                            : cachedBuffers.peek().f0.buffer().getDataType(),
                    sequenceNumber++);
        }
    }
@Test
    void testGetNextBuffer() throws IOException {
        assertThat(view.peekNextBufferSubpartitionId()).isEqualTo(-1);
        assertThat(view.getNextBuffer()).isNull();
        view0.notifyDataAvailable();
        assertThat(view.peekNextBufferSubpartitionId()).isZero();
        ResultSubpartition.BufferAndBacklog bufferAndBacklog = view.getNextBuffer();
        assertThat(bufferAndBacklog.buffer()).isEqualTo(buffers0.get(0));
        assertThat(bufferAndBacklog.buffersInBacklog()).isEqualTo(buffers0.size() - 1);

        view1.notifyDataAvailable();
        assertThat(view.peekNextBufferSubpartitionId()).isZero();
        assertThat(view.getNextBuffer().buffer()).isEqualTo(buffers0.get(1));

        List<Buffer> buffers = new ArrayList<>();
        while (view.getAvailabilityAndBacklog(true).isAvailable()) {
            buffers.add(view.getNextBuffer().buffer());
        }
        assertThat(buffers)
                .hasSize(buffers0.size() + buffers1.size() - 2)
                .containsSubsequence(buffers0.subList(2, buffers0.size()))
                .containsSubsequence(buffers1);
    }