@Override
    public Optional<Buffer> getNextBuffer(
            TieredStoragePartitionId partitionId,
            TieredStorageSubpartitionId subpartitionId,
            int segmentId) {
        // Get current segment id and buffer index.
        Tuple2<Integer, Integer> bufferIndexAndSegmentId =
                currentBufferIndexAndSegmentIds
                        .computeIfAbsent(partitionId, ignore -> new HashMap<>())
                        .getOrDefault(subpartitionId, Tuple2.of(0, 0));
        int currentBufferIndex = bufferIndexAndSegmentId.f0;
        int currentSegmentId = bufferIndexAndSegmentId.f1;
        if (segmentId != currentSegmentId) {
            remoteStorageScanner.watchSegment(partitionId, subpartitionId, segmentId);
        }

        // Read buffer from the partition file in remote storage.
        MemorySegment memorySegment = MemorySegmentFactory.allocateUnpooledSegment(bufferSizeBytes);
        PartitionFileReader.ReadBufferResult readBufferResult = null;
        try {
            readBufferResult =
                    partitionFileReader.readBuffer(
                            partitionId,
                            subpartitionId,
                            segmentId,
                            currentBufferIndex,
                            memorySegment,
                            FreeingBufferRecycler.INSTANCE,
                            null,
                            null);
        } catch (IOException e) {
            memorySegment.free();
            ExceptionUtils.rethrow(e, "Failed to read buffer from partition file.");
        }
        if (readBufferResult != null && !readBufferResult.getReadBuffers().isEmpty()) {
            List<Buffer> readBuffers = readBufferResult.getReadBuffers();
            checkState(readBuffers.size() == 1);
            Buffer buffer = readBuffers.get(0);
            currentBufferIndexAndSegmentIds
                    .get(partitionId)
                    .put(subpartitionId, Tuple2.of(++currentBufferIndex, segmentId));
            return Optional.of(buffer);
        } else {
            memorySegment.free();
        }
        synchronized (availableSubpartitionsQueues) {
            availableSubpartitionsQueues.get(partitionId).remove(subpartitionId);
        }
        return Optional.empty();
    }
@Test
    void testGetEmptyBuffer() {
        TieredStoragePartitionId partitionId =
                new TieredStoragePartitionId(new ResultPartitionID());
        RemoteTierConsumerAgent remoteTierConsumerAgent =
                new RemoteTierConsumerAgent(
                        Collections.singletonList(
                                new TieredStorageConsumerSpec(
                                        0,
                                        partitionId,
                                        new TieredStorageInputChannelId(0),
                                        new ResultSubpartitionIndexSet(0))),
                        new RemoteStorageScanner(remoteStoragePath),
                        new TestingPartitionFileReader.Builder().build(),
                        1024);
        assertThat(
                        remoteTierConsumerAgent.getNextBuffer(
                                partitionId, new TieredStorageSubpartitionId(0), 0))
                .isEmpty();
    }