private void getBlocks0(GetBlocksContext ctx, long startOffset, long endOffset, int maxBytes) {
        Long floorKey = blocksMap.floorKey(startOffset);
        if (floorKey == null && !blocksMap.isEmpty()) {
            if (ctx.readahead) {
                // The user read may be faster than the readahead, and clear the read completed blocks.
                ctx.cf.complete(ctx.blocks);
                return;
            } else {
                LOGGER.error("[BUG] {} cannot find floor block for startOffset={}, the first block={}", this, startOffset, blocksMap.firstEntry());
                throw new AutoMQException("[BUG] cannot find floor block");
            }
        }
        CompletableFuture<Boolean> loadMoreBlocksCf;
        int remainingSize = maxBytes;
        if (floorKey == null || startOffset >= loadedBlockIndexEndOffset) {
            loadMoreBlocksCf = loadMoreBlocksWithoutData(endOffset);
        } else {
            boolean firstBlock = true;
            boolean fulfill = false;
            for (Map.Entry<Long, Block> entry : blocksMap.tailMap(floorKey).entrySet()) {
                Block block = entry.getValue();
                long objectId = block.metadata.objectId();
                if (!objectManager.isObjectExist(objectId)) {
                    // The cached block's object maybe deleted by the compaction. So we need to check the object exist.
                    ctx.cf.completeExceptionally(new ObjectNotExistException(objectId));
                    return;
                }
                DataBlockIndex index = block.index;
                if (!firstBlock || index.startOffset() == startOffset) {
                    remainingSize -= index.size();
                }
                if (firstBlock) {
                    firstBlock = false;
                }
                // after read the data will be return to the cache, so we need to reload the data every time
                block = block.newBlockWithData(ctx.readahead);
                ctx.blocks.add(block);
                if ((endOffset != -1L && index.endOffset() >= endOffset) || remainingSize <= 0) {
                    fulfill = true;
                    break;
                }
            }
            if (fulfill) {
                ctx.cf.complete(ctx.blocks);
                return;
            } else {
                loadMoreBlocksCf = loadMoreBlocksWithoutData(endOffset);
            }
        }
        int finalRemainingSize = remainingSize;
        // use async to prevent recursive call cause stack overflow
        loadMoreBlocksCf.thenAcceptAsync(moreBlocks -> {
            if (ctx.readahead) {
                // If #loadMoreBlocksWithoutData result is empty, it means the stream is already loads to the end.
                if (!moreBlocks) {
                    ctx.cf.complete(ctx.blocks);
                    return;
                }
            } else {
                if (!moreBlocks && endOffset > loadedBlockIndexEndOffset) {
                    String errMsg = String.format("[BUG] streamId=%s expect load blocks to endOffset=%s, " + "current loadedBlockIndexEndOffset=%s", streamId, endOffset, loadedBlockIndexEndOffset);
                    ctx.cf.completeExceptionally(new AutoMQException(errMsg));
                    return;
                }
            }
            long nextStartOffset = ctx.blocks.isEmpty() ? startOffset : ctx.blocks.get(ctx.blocks.size() - 1).index.endOffset();
            getBlocks0(ctx, nextStartOffset, endOffset, finalRemainingSize);
        }, eventLoop).exceptionally(ex -> {
            ctx.cf.completeExceptionally(ex);
            return null;
        });
    }
This method is only for unit testing.
inflightReadaheadCf is empty, the task has been completed
inflightReadaheadCf is not empty, the task may be completed

@return readahead.inflightReadaheadCf
s3stream/src/main/java/com/automq/stream/s3/cache/blockcache/StreamReader.java