void afterRead(ReadDataBlock readDataBlock, ReadContext ctx) {
        List<StreamRecordBatch> records = readDataBlock.getRecords();
        if (!records.isEmpty()) {
            nextReadOffset = records.get(records.size() - 1).getLastOffset();
        }
        // clear unused blocks
        Iterator<Map.Entry<Long, Block>> it = blocksMap.entrySet().iterator();
        while (it.hasNext()) {
            Block block = it.next().getValue();
            if (block.index.endOffset() <= nextReadOffset) {
                it.remove();
                block.markReadCompleted();
            } else {
                break;
            }
        }
        for (Block block : ctx.blocks) {
            block.release();
        }
        // try readahead to speed up the next read
        afterReadTryReadaheadCf = eventLoop.submit(() -> readahead.tryReadahead(readDataBlock.getCacheAccessType() == BLOCK_CACHE_MISS));
        afterReadTryReadaheadCf.whenComplete((nil, ex) -> {
            Throwable cause = FutureUtil.cause(ex);
            if (cause != null && !isRecoverable(cause)) {
                LOGGER.error("AfterRead failed", ex);
            }
            // help gc
            afterReadTryReadaheadCf = 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