public ValidationResult validateMessagesAndAssignOffsets(PrimitiveRef.LongRef offsetCounter,
                                                             MetricsRecorder metricsRecorder,
                                                             BufferSupplier bufferSupplier) {
        if (sourceCompressionType == CompressionType.NONE && targetCompression.type() == CompressionType.NONE) {
            // check the magic value
            if (!records.hasMatchingMagic(toMagic))
                return convertAndAssignOffsetsNonCompressed(offsetCounter, metricsRecorder);
            else
                // Do in-place validation, offset assignment and maybe set timestamp
                return assignOffsetsNonCompressed(offsetCounter, metricsRecorder);
        } else
            return validateMessagesAndAssignOffsetsCompressed(offsetCounter, metricsRecorder, bufferSupplier);

    }
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