public Block newBlockWithData(boolean readahead) {
            // We need to create a new block with consistent data to avoid duplicated release or leak,
            // cause of the loaded data maybe evicted and reloaded.
            Block newBlock = new Block(metadata, index);
            ObjectReader objectReader = objectReaderFactory.get(metadata);
            DataBlockCache.GetOptions getOptions = DataBlockCache.GetOptions.builder().readahead(readahead).build();
            loadCf = dataBlockCache.getBlock(getOptions, objectReader, index).thenAccept(newData -> {
                newBlock.data = newData;
                if (!readCompleted && data != newData) {
                    // the data block is first loaded or evict & reload
                    if (data != null) {
                        freeListenerHandle.close();
                    }
                    data = newData;
                    newData.markUnread();
                    freeListenerHandle = data.registerFreeListener(b -> handleBlockFree(this));
                }
            }).exceptionally(ex -> {
                exception = ex;
                newBlock.exception = ex;
                return null;
            }).whenComplete((nil, ex) -> objectReader.release());
            newBlock.loadCf = loadCf;
            return newBlock;
        }
Put block into the blocks

@param block {@link Block}
@return if the block is continuous to the last block, it will return true
s3stream/src/main/java/com/automq/stream/s3/cache/blockcache/StreamReader.java