@Override
    public Buffer.DataType peekNextToConsumeDataType(
            int nextBufferToConsume, Collection<Buffer> buffersToRecycle) {
        Buffer.DataType dataType = Buffer.DataType.NONE;
        try {
            dataType =
                    checkAndGetFirstBufferIndexOrError(nextBufferToConsume, buffersToRecycle)
                            .map(BufferIndexOrError::getDataType)
                            .orElse(Buffer.DataType.NONE);
        } catch (Throwable throwable) {
            ExceptionUtils.rethrow(throwable);
        }
        return dataType;
    }
@Test
    void testPeekNextToConsumeDataType() throws Throwable {
        TestingSubpartitionConsumerInternalOperation viewNotifier =
                new TestingSubpartitionConsumerInternalOperation();
        HsSubpartitionFileReaderImpl subpartitionFileReader =
                createSubpartitionFileReader(0, viewNotifier);

        // if no preload data in file reader, return DataType.NONE.
        assertThat(subpartitionFileReader.peekNextToConsumeDataType(0, new ArrayList<>()))
                .isEqualTo(DataType.NONE);

        // buffers in file: (0-0, 0-1, 0-2)
        writeDataToFile(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, return the real DataType.
        assertThat(subpartitionFileReader.peekNextToConsumeDataType(0, Collections.emptyList()))
                .isEqualTo(DataType.DATA_BUFFER);

        // if nextBufferToConsume is greater than peek elements index, skip this buffer and keep
        // looking.
        assertThat(subpartitionFileReader.peekNextToConsumeDataType(2, new ArrayList<>()))
                .isEqualTo(DataType.EVENT_BUFFER);
        // if nextBufferToConsume is less than peek elements index, return DataType.NONE.
        assertThat(subpartitionFileReader.peekNextToConsumeDataType(1, Collections.emptyList()))
                .isEqualTo(DataType.NONE);
    }