@Override
        protected Record makeNext() {
            if (innerEntries.isEmpty())
                return allDone();

            AbstractLegacyRecordBatch entry = innerEntries.remove();

            // Convert offset to absolute offset if needed.
            if (wrapperMagic == RecordBatch.MAGIC_VALUE_V1) {
                long absoluteOffset = absoluteBaseOffset + entry.offset();
                entry = new BasicLegacyRecordBatch(absoluteOffset, entry.outerRecord());
            }

            if (entry.isCompressed())
                throw new InvalidRecordException("Inner messages must not be compressed");

            return entry;
        }
Get an iterator for the nested entries contained within this batch. Note that
if the batch is not compressed, then this method will return an iterator over the
shallow record only (i.e. this object).
@return An iterator over the records contained within this batch
clients/src/main/java/org/apache/kafka/common/record/AbstractLegacyRecordBatch.java