@Override
        protected ConvertedRecords<?> makeNext() {
            // If we have cached the first down-converted batch, return that now
            if (firstConvertedBatch != null) {
                ConvertedRecords<?> convertedBatch = firstConvertedBatch;
                firstConvertedBatch = null;
                return convertedBatch;
            }

            while (batchIterator.hasNext()) {
                final List<RecordBatch> batches = new ArrayList<>();
                boolean isFirstBatch = true;
                long sizeSoFar = 0;

                // Figure out batches we should down-convert based on the size constraints
                while (batchIterator.hasNext() &&
                        (isFirstBatch || (batchIterator.peek().sizeInBytes() + sizeSoFar) <= maximumReadSize)) {
                    RecordBatch currentBatch = batchIterator.next();
                    batches.add(currentBatch);
                    sizeSoFar += currentBatch.sizeInBytes();
                    isFirstBatch = false;
                }

                ConvertedRecords<MemoryRecords> convertedRecords = RecordsUtil.downConvert(batches, toMagic, firstOffset, time);
                // During conversion, it is possible that we drop certain batches because they do not have an equivalent
                // representation in the message format we want to convert to. For example, V0 and V1 message formats
                // have no notion of transaction markers which were introduced in V2 so they get dropped during conversion.
                // We return converted records only when we have at least one valid batch of messages after conversion.
                if (convertedRecords.records().sizeInBytes() > 0)
                    return convertedRecords;
            }
            return allDone();
        }
Make next set of down-converted records
@return Down-converted records
clients/src/main/java/org/apache/kafka/common/record/LazyDownConversionRecords.java