@Override
    public Optional<ResultSubpartition.BufferAndBacklog> consumeBuffer(
            int nextBufferToConsume, Collection<Buffer> buffersToRecycle) throws Throwable {
        if (!checkAndGetFirstBufferIndexOrError(nextBufferToConsume, buffersToRecycle)
                .isPresent()) {
            return Optional.empty();
        }

        // already ensure that peek element is not null and not throwable.
        BufferIndexOrError current = checkNotNull(loadedBuffers.poll());

        BufferIndexOrError next = loadedBuffers.peek();

        Buffer.DataType nextDataType = next == null ? Buffer.DataType.NONE : next.getDataType();
        int bufferIndex = current.getIndex();
        Buffer buffer =
                current.getBuffer()
                        .orElseThrow(
                                () ->
                                        new NullPointerException(
                                                "Get a non-throwable and non-buffer bufferIndexOrError, which is not allowed"));
        tryDecreaseBacklog(buffer);
        return Optional.of(
                ResultSubpartition.BufferAndBacklog.fromBufferAndLookahead(
                        buffer, nextDataType, backlog.get(), bufferIndex));
    }
@Test
    void testConsumeBuffer() throws Throwable {
        TestingSubpartitionConsumerInternalOperation viewNotifier =
                new TestingSubpartitionConsumerInternalOperation();
        HsSubpartitionFileReaderImpl subpartitionFileReader =
                createSubpartitionFileReader(0, viewNotifier);

        // if no preload data in file reader, return Optional.empty.
        assertThat(subpartitionFileReader.consumeBuffer(0, Collections.emptyList())).isNotPresent();

        // buffers in file: (0-0, 0-1, 0-2)
        writeDataToFile(0, 0, 0, 3);

        Queue<MemorySegment> memorySegments = createsMemorySegments(3);
        subpartitionFileReader.prepareForScheduling();
        // trigger reading, add buffer to queue.
        subpartitionFileReader.readBuffers(memorySegments, (ignore) -> {});

        // if nextBufferToConsume is equal to peek elements index.
        assertThat(subpartitionFileReader.consumeBuffer(0, new ArrayList<>()))
                .hasValueSatisfying(
                        (bufferAndBacklog -> {
                            assertThat(bufferAndBacklog.getNextDataType())
                                    .isEqualTo(DataType.DATA_BUFFER);
                            assertThat(bufferAndBacklog.getSequenceNumber()).isEqualTo(0);
                            // first buffer's data is 0.
                            assertThat(
                                            bufferAndBacklog
                                                    .buffer()
                                                    .getNioBufferReadable()
                                                    .order(ByteOrder.nativeOrder())
                                                    .getInt())
                                    .isEqualTo(0);
                        }));

        // if nextBufferToConsume is less than peek elements index, return Optional.empty.
        assertThat(subpartitionFileReader.consumeBuffer(0, Collections.emptyList())).isNotPresent();

        // if nextBufferToConsume is greater than peek elements index, skip this buffer and keep
        // looking.
        assertThat(subpartitionFileReader.consumeBuffer(2, new ArrayList<>()))
                .hasValueSatisfying(
                        (bufferAndBacklog -> {
                            assertThat(bufferAndBacklog.getNextDataType()).isEqualTo(DataType.NONE);
                            assertThat(bufferAndBacklog.getSequenceNumber()).isEqualTo(2);
                            assertThat(
                                            bufferAndBacklog
                                                    .buffer()
                                                    .getNioBufferReadable()
                                                    .order(ByteOrder.nativeOrder())
                                                    .getInt())
                                    .isEqualTo(2);
                        }));
        assertThat(subpartitionFileReader.getLoadedBuffers()).isEmpty();
    }