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 testOffsetAssignmentAfterUpConversionV1ToV2Compressed() {
        Compression compression = Compression.gzip().build();
        MemoryRecords records = createRecords(RecordBatch.MAGIC_VALUE_V1, RecordBatch.NO_TIMESTAMP, compression);
        long offset = 1234567;
        checkOffsets(records, 0);
        checkOffsets(new LogValidator(
                records,
                new TopicPartition("topic", 0),
                time,
                CompressionType.GZIP,
                compression,
                false,
                RecordBatch.MAGIC_VALUE_V2,
                TimestampType.LOG_APPEND_TIME,
                1000L,
                1000L,
                RecordBatch.NO_PARTITION_LEADER_EPOCH,
                AppendOrigin.CLIENT,
                MetadataVersion.latestTesting()
        ).validateMessagesAndAssignOffsets(
                PrimitiveRef.ofLong(offset), metricsRecorder, RequestLocal.withThreadConfinedCaching().bufferSupplier()
        ).validatedRecords, offset);
    }