@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 testReadProgress() throws IOException {
        long currentFileOffset = 0;
        ProducerMergedPartitionFileReader.ProducerMergedReadProgress readProgress = null;
        CompositeBuffer partialBuffer = null;
        for (int bufferIndex = 0; bufferIndex < DEFAULT_BUFFER_NUMBER; ) {
            PartitionFileReader.ReadBufferResult readBufferResult =
                    readBuffer(bufferIndex, DEFAULT_SUBPARTITION_ID, readProgress, partialBuffer);
            assertThat(readBufferResult).isNotNull();
            assertThat(readBufferResult.getReadProgress())
                    .isInstanceOf(
                            ProducerMergedPartitionFileReader.ProducerMergedReadProgress.class);
            readProgress =
                    (ProducerMergedPartitionFileReader.ProducerMergedReadProgress)
                            readBufferResult.getReadProgress();
            for (Buffer buffer : readBufferResult.getReadBuffers()) {
                if (buffer instanceof CompositeBuffer) {
                    partialBuffer = (CompositeBuffer) buffer;
                    if (partialBuffer.missingLength() == 0) {
                        bufferIndex++;
                        currentFileOffset += partialBuffer.readableBytes() + HEADER_LENGTH;
                        partialBuffer.recycleBuffer();
                        partialBuffer = null;
                    }
                } else {
                    bufferIndex++;
                    currentFileOffset += buffer.readableBytes() + HEADER_LENGTH;
                    buffer.recycleBuffer();
                }
            }
            assertThat(readProgress.getCurrentBufferOffset()).isEqualTo(currentFileOffset);
        }
    }