@Override
    public ReadBufferResult readBuffer(
            TieredStoragePartitionId partitionId,
            TieredStorageSubpartitionId subpartitionId,
            int segmentId,
            int bufferIndex,
            MemorySegment memorySegment,
            BufferRecycler recycler,
            @Nullable ReadProgress readProgress,
            @Nullable CompositeBuffer partialBuffer)
            throws IOException {

        lazyInitializeFileChannel();

        // Get the read offset, including the start offset, the end offset
        Tuple2<Long, Long> startAndEndOffset =
                getReadStartAndEndOffset(subpartitionId, bufferIndex, readProgress, partialBuffer);
        if (startAndEndOffset == null) {
            return null;
        }
        long readStartOffset = startAndEndOffset.f0;
        long readEndOffset = startAndEndOffset.f1;

        int numBytesToRead =
                Math.min(memorySegment.size(), (int) (readEndOffset - readStartOffset));

        if (numBytesToRead == 0) {
            return null;
        }

        List<Buffer> readBuffers = new LinkedList<>();
        ByteBuffer byteBuffer = memorySegment.wrap(0, numBytesToRead);
        fileChannel.position(readStartOffset);
        // Read data to the memory segment, note the read size is numBytesToRead
        readFileDataToBuffer(memorySegment, recycler, byteBuffer);

        // Slice the read memory segment to multiple small network buffers and add them to
        // readBuffers
        Tuple2<Integer, Integer> partial =
                sliceBuffer(byteBuffer, memorySegment, partialBuffer, recycler, readBuffers);

        return getReadBufferResult(
                readBuffers,
                readStartOffset,
                readEndOffset,
                numBytesToRead,
                partial.f0,
                partial.f1);
    }
@Test
    void testReadBuffer() throws IOException {
        for (int bufferIndex = 0; bufferIndex < DEFAULT_BUFFER_NUMBER; ++bufferIndex) {
            List<Buffer> buffers = readBuffer(bufferIndex, DEFAULT_SUBPARTITION_ID);
            assertThat(buffers).isNotNull();
            buffers.forEach(Buffer::recycleBuffer);
        }
        MemorySegment memorySegment =
                MemorySegmentFactory.allocateUnpooledSegment(DEFAULT_BUFFER_SIZE);
        assertThat(
                        partitionFileReader.readBuffer(
                                DEFAULT_PARTITION_ID,
                                DEFAULT_SUBPARTITION_ID,
                                DEFAULT_SEGMENT_ID,
                                DEFAULT_BUFFER_NUMBER + 1,
                                memorySegment,
                                FreeingBufferRecycler.INSTANCE,
                                null,
                                null))
                .isNull();
    }