private ValidationResult convertAndAssignOffsetsNonCompressed(LongRef offsetCounter,
                                                                  MetricsRecorder metricsRecorder) {
        long now = time.milliseconds();
        long startNanos = time.nanoseconds();
        int sizeInBytesAfterConversion = AbstractRecords.estimateSizeInBytes(toMagic, offsetCounter.value,
            CompressionType.NONE, records.records());

        RecordBatch firstBatch = getFirstBatchAndMaybeValidateNoMoreBatches(records, CompressionType.NONE);
        long producerId = firstBatch.producerId();
        short producerEpoch = firstBatch.producerEpoch();
        int sequence = firstBatch.baseSequence();
        boolean isTransactional = firstBatch.isTransactional();

        // The current implementation of BufferSupplier is naive and works best when the buffer size
        // cardinality is low, so don't use it here
        ByteBuffer newBuffer = ByteBuffer.allocate(sizeInBytesAfterConversion);
        MemoryRecordsBuilder builder = MemoryRecords.builder(newBuffer, toMagic, Compression.NONE,
            timestampType, offsetCounter.value, now, producerId, producerEpoch, sequence, isTransactional,
            partitionLeaderEpoch);

        for (RecordBatch batch : records.batches()) {
            validateBatch(topicPartition, firstBatch, batch, origin, toMagic, metricsRecorder);

            List<ApiRecordError> recordErrors = new ArrayList<>(0);
            int recordIndex = 0;
            for (Record record : batch) {
                Optional<ApiRecordError> recordError = validateRecord(batch, topicPartition,
                    record, recordIndex, now, timestampType, timestampBeforeMaxMs, timestampAfterMaxMs, compactedTopic,
                    metricsRecorder);
                recordError.ifPresent(e -> recordErrors.add(e));
                // we fail the batch if any record fails, so we stop appending if any record fails
                if (recordErrors.isEmpty())
                    builder.appendWithOffset(offsetCounter.value++, record);
                ++recordIndex;
            }

            processRecordErrors(recordErrors);
        }

        MemoryRecords convertedRecords = builder.build();

        RecordsInfo info = builder.info();
        RecordValidationStats recordValidationStats = new RecordValidationStats(
            builder.uncompressedBytesWritten(), builder.numRecords(), time.nanoseconds() - startNanos);
        return new ValidationResult(
            now,
            convertedRecords,
            info.maxTimestamp,
            info.shallowOffsetOfMaxTimestamp,
            true,
            recordValidationStats);
    }
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