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 testRecordWithPastTimestampIsRejected() {
        long timestampBeforeMaxConfig = Duration.ofHours(24).toMillis(); // 24 hrs
        long timestampAfterMaxConfig = Duration.ofHours(1).toMillis(); // 1 hr
        long now = System.currentTimeMillis();
        long fiveMinutesBeforeThreshold = now - timestampBeforeMaxConfig - Duration.ofMinutes(5).toMillis();
        Compression compression = Compression.gzip().build();
        MemoryRecords records = createRecords(RecordBatch.MAGIC_VALUE_V2, fiveMinutesBeforeThreshold, compression);
        RecordValidationException e = assertThrows(RecordValidationException.class, () ->
                new LogValidator(
                        records,
                        topicPartition,
                        time,
                        CompressionType.GZIP,
                        compression,
                        false,
                        RecordBatch.MAGIC_VALUE_V2,
                        TimestampType.CREATE_TIME,
                        timestampBeforeMaxConfig,
                        timestampAfterMaxConfig,
                        RecordBatch.NO_PARTITION_LEADER_EPOCH,
                        AppendOrigin.CLIENT,
                        MetadataVersion.latestTesting()
                ).validateMessagesAndAssignOffsets(
                        PrimitiveRef.ofLong(0L), metricsRecorder, RequestLocal.withThreadConfinedCaching().bufferSupplier()
                )
        );

        assertInstanceOf(InvalidTimestampException.class, e.invalidException());
        assertFalse(e.recordErrors().isEmpty());
        assertEquals(3, e.recordErrors().size());
    }