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 testDifferentLevelDoesNotCauseRecompression() {
        List<byte[]> records = Arrays.asList(
                String.join("", Collections.nCopies(256, "some")).getBytes(),
                String.join("", Collections.nCopies(256, "data")).getBytes()
        );

        // Records from the producer were created with gzip max level
        Compression gzipMax = Compression.gzip().level(CompressionType.GZIP.maxLevel()).build();
        MemoryRecords recordsGzipMax = createRecords(records, RecordBatch.MAGIC_VALUE_V2, RecordBatch.NO_TIMESTAMP, gzipMax);

        // The topic is configured with gzip min level
        Compression gzipMin = Compression.gzip().level(CompressionType.GZIP.minLevel()).build();
        MemoryRecords recordsGzipMin = createRecords(records, RecordBatch.MAGIC_VALUE_V2, RecordBatch.NO_TIMESTAMP, gzipMin);

        // Ensure data compressed with gzip max and min is different
        assertNotEquals(recordsGzipMax, recordsGzipMin);

        LogValidator validator = new LogValidator(recordsGzipMax,
                topicPartition,
                time,
                gzipMax.type(),
                gzipMin,
                false,
                RecordBatch.MAGIC_VALUE_V2,
                TimestampType.CREATE_TIME,
                5000L,
                5000L,
                RecordBatch.NO_PARTITION_LEADER_EPOCH,
                AppendOrigin.CLIENT,
                MetadataVersion.latestTesting()
        );

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

        // Ensure validated records have not been changed so they are the same as the producer records
        assertEquals(recordsGzipMax, result.validatedRecords);
        assertNotEquals(recordsGzipMin, result.validatedRecords);
    }