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

        // Get the channel of the segment file for a subpartition.
        Map<TieredStorageSubpartitionId, Tuple2<ReadableByteChannel, Integer>> subpartitionInfo =
                openedChannelAndSegmentIds.computeIfAbsent(partitionId, ignore -> new HashMap<>());
        Tuple2<ReadableByteChannel, Integer> fileChannelAndSegmentId =
                subpartitionInfo.getOrDefault(subpartitionId, Tuple2.of(null, -1));
        ReadableByteChannel channel = fileChannelAndSegmentId.f0;

        // Create the channel if there is a new segment file for a subpartition.
        if (channel == null || fileChannelAndSegmentId.f1 != segmentId) {
            if (channel != null) {
                channel.close();
            }
            channel = openNewChannel(partitionId, subpartitionId, segmentId);
            if (channel == null) {
                // return null if the segment file doesn't exist.
                return null;
            }
            subpartitionInfo.put(subpartitionId, Tuple2.of(channel, segmentId));
        }

        // Try to read a buffer from the channel.
        reusedHeaderBuffer.clear();
        int bufferHeaderResult = channel.read(reusedHeaderBuffer);
        if (bufferHeaderResult == -1) {
            channel.close();
            openedChannelAndSegmentIds.get(partitionId).remove(subpartitionId);
            return getSingletonReadResult(
                    new NetworkBuffer(memorySegment, recycler, Buffer.DataType.END_OF_SEGMENT));
        }
        reusedHeaderBuffer.flip();
        BufferHeader header = parseBufferHeader(reusedHeaderBuffer);
        int dataBufferResult = channel.read(memorySegment.wrap(0, header.getLength()));
        if (dataBufferResult != header.getLength()) {
            channel.close();
            throw new IOException("The length of data buffer is illegal.");
        }
        Buffer.DataType dataType = header.getDataType();
        return getSingletonReadResult(
                new NetworkBuffer(
                        memorySegment,
                        recycler,
                        dataType,
                        header.isCompressed(),
                        header.getLength()));
    }
@Test
    void testReadBuffer() throws IOException {
        for (int subpartitionId = 0; subpartitionId < DEFAULT_NUM_SUBPARTITION; ++subpartitionId) {
            for (int segmentId = 0; segmentId < DEFAULT_SEGMENT_NUM; ++segmentId) {
                for (int bufferIndex = 0; bufferIndex < DEFAULT_BUFFER_PER_SEGMENT; ++bufferIndex) {
                    Buffer buffer =
                            readBuffer(
                                    bufferIndex,
                                    new TieredStorageSubpartitionId(subpartitionId),
                                    segmentId);
                    assertThat(buffer).isNotNull();
                    buffer.recycleBuffer();
                }
            }
        }
    }