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 testNonIncreasingOffsetRecordBatchHasMetricsLogged() {
        ByteBuffer buf = ByteBuffer.allocate(512);
        MemoryRecordsBuilder builder = MemoryRecords.builder(buf, RecordBatch.MAGIC_VALUE_V2, Compression.NONE, TimestampType.CREATE_TIME, 0L);
        builder.appendWithOffset(0, RecordBatch.NO_TIMESTAMP, null, "hello".getBytes());
        builder.appendWithOffset(2, RecordBatch.NO_TIMESTAMP, null, "there".getBytes());
        builder.appendWithOffset(3, RecordBatch.NO_TIMESTAMP, null, "beautiful".getBytes());

        MemoryRecords records = builder.build();
        records.batches().iterator().next().setLastOffset(2);
        assertThrows(InvalidRecordException.class, () -> new LogValidator(
                records,
                new TopicPartition("topic", 0),
                time,
                CompressionType.GZIP,
                Compression.gzip().build(),
                false,
                RecordBatch.MAGIC_VALUE_V0,
                TimestampType.CREATE_TIME,
                5000L,
                5000L,
                RecordBatch.NO_PARTITION_LEADER_EPOCH,
                AppendOrigin.CLIENT,
                MetadataVersion.latestTesting()
        ).validateMessagesAndAssignOffsets(
                PrimitiveRef.ofLong(0L), metricsRecorder, RequestLocal.withThreadConfinedCaching().bufferSupplier()
        ));

        assertEquals(metricsRecorder.recordInvalidOffsetCount, 1);
    }