private static MutableRecordBatch getFirstBatchAndMaybeValidateNoMoreBatches(MemoryRecords records,
                                                                                 CompressionType sourceCompression) {
        Iterator<MutableRecordBatch> batchIterator = records.batches().iterator();

        if (!batchIterator.hasNext())
            throw new InvalidRecordException("Record batch has no batches at all");

        MutableRecordBatch batch = batchIterator.next();

        // if the format is v2 and beyond, or if the messages are compressed, we should check there's only one batch.
        if (batch.magic() >= RecordBatch.MAGIC_VALUE_V2 || sourceCompression != CompressionType.NONE) {
            if (batchIterator.hasNext())
                throw new InvalidRecordException("Compressed outer record has more than one batch");
        }

        return batch;
    }
Update the offsets for this message set and do further validation on messages including:
1. Messages for compacted topics must have keys
2. When magic value >= 1, inner messages of a compressed message set must have monotonically increasing offsets
   starting from 0.
3. When magic value >= 1, validate and maybe overwrite timestamps of messages.
4. Declared count of records in DefaultRecordBatch must match number of valid records contained therein.

This method will convert messages as necessary to the topic's configured message format version. If no format
conversion or value overwriting is required for messages, this method will perform in-place operations to
avoid expensive re-compression.

Returns a ValidationAndOffsetAssignResult containing the validated message set, maximum timestamp, the offset
of the message with the max timestamp and a boolean indicating whether the message sizes may have changed.
storage/src/main/java/org/apache/kafka/storage/internals/log/LogValidator.java