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 testControlRecordsNotAllowedFromClients() {
        long offset = 1234567;
        EndTransactionMarker endTxnMarker = new EndTransactionMarker(ControlRecordType.COMMIT, 0);
        MemoryRecords records = MemoryRecords.withEndTransactionMarker(23423L, (short) 5, endTxnMarker);
        assertThrows(InvalidRecordException.class, () -> new LogValidator(
                records,
                new TopicPartition("topic", 0),
                time,
                CompressionType.NONE,
                Compression.NONE,
                false,
                RecordBatch.CURRENT_MAGIC_VALUE,
                TimestampType.CREATE_TIME,
                5000L,
                5000L,
                RecordBatch.NO_PARTITION_LEADER_EPOCH,
                AppendOrigin.CLIENT,
                MetadataVersion.latestTesting()
        ).validateMessagesAndAssignOffsets(
                PrimitiveRef.ofLong(offset),
                metricsRecorder,
                RequestLocal.withThreadConfinedCaching().bufferSupplier()
        ));
    }