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);

    }
@Test
    public void testCreateTimeUpConversionV1ToV2() {
        long timestamp = System.currentTimeMillis();
        Compression compression = Compression.gzip().build();
        MemoryRecords records = createRecords(RecordBatch.MAGIC_VALUE_V1, timestamp, compression);

        LogValidator validator = new LogValidator(
                records,
                topicPartition,
                time,
                CompressionType.GZIP,
                compression,
                false,
                RecordBatch.MAGIC_VALUE_V2,
                TimestampType.CREATE_TIME,
                1000L,
                1000L,
                RecordBatch.NO_PARTITION_LEADER_EPOCH,
                AppendOrigin.CLIENT,
                MetadataVersion.latestTesting()
        );

        LogValidator.ValidationResult validatedResults = validator.validateMessagesAndAssignOffsets(
                PrimitiveRef.ofLong(0),
                metricsRecorder,
                RequestLocal.withThreadConfinedCaching().bufferSupplier()
        );

        MemoryRecords validatedRecords = validatedResults.validatedRecords;

        for (RecordBatch batch : validatedRecords.batches()) {
            assertTrue(batch.isValid());
            maybeCheckBaseTimestamp(timestamp, batch);
            assertEquals(timestamp, batch.maxTimestamp());
            assertEquals(TimestampType.CREATE_TIME, batch.timestampType());
            assertEquals(RecordBatch.NO_PRODUCER_EPOCH, batch.producerEpoch());
            assertEquals(RecordBatch.NO_PRODUCER_ID, batch.producerId());
            assertEquals(RecordBatch.NO_SEQUENCE, batch.baseSequence());
        }

        assertEquals(timestamp, validatedResults.maxTimestampMs);
        assertEquals(2, validatedResults.shallowOffsetOfMaxTimestamp, "Offset of max timestamp should be the last offset 2.");
        assertTrue(validatedResults.messageSizeMaybeChanged, "Message size should have been changed");

        verifyRecordValidationStats(
                validatedResults.recordValidationStats,
                3,
                records,
                true
        );
    }