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 testDifferentCodecCausesRecompression() {
        List<byte[]> records = Arrays.asList(
                "somedata".getBytes(),
                "moredata".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 lz4 min level
        Compression lz4Min = Compression.lz4().level(CompressionType.GZIP.minLevel()).build();
        MemoryRecords recordsLz4Min = createRecords(records, RecordBatch.MAGIC_VALUE_V2, RecordBatch.NO_TIMESTAMP, lz4Min);

        LogValidator validator = new LogValidator(recordsGzipMax,
                topicPartition,
                time,
                gzipMax.type(),
                lz4Min,
                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 been recompressed and match lz4 min level
        assertEquals(recordsLz4Min, result.validatedRecords);
    }