public CloseableIterator<StreamRecordBatch> iterator(boolean copy) {
            ByteBuf buf = this.buf.duplicate();
            AtomicInteger currentBlockRecordCount = new AtomicInteger(0);
            AtomicInteger remainingRecordCount = new AtomicInteger(recordCount);
            return new CloseableIterator<>() {
                @Override
                public boolean hasNext() {
                    // in.available() is not reliable. ZstdInputStreamNoFinalizer might return 1 when there is no more data.
                    return remainingRecordCount.get() != 0;
                }

                @Override
                public StreamRecordBatch next() {
                    if (remainingRecordCount.decrementAndGet() < 0) {
                        throw new NoSuchElementException();
                    }
                    if (currentBlockRecordCount.get() == 0) {
                        buf.skipBytes(1 /* magic */ + 1 /* flag */);
                        currentBlockRecordCount.set(buf.readInt());
                        buf.skipBytes(4);
                    }
                    currentBlockRecordCount.decrementAndGet();
                    return copy ? StreamRecordBatchCodec.duplicateDecode(buf) : StreamRecordBatchCodec.sliceRetainDecode(buf);
                }

                @Override
                public void close() {
                }
            };
        }
@param dataBlockSize The total size of the data blocks, which equals to index start position.
@param indexBlock    raw index data.
s3stream/src/main/java/com/automq/stream/s3/ObjectReader.java