@ParameterizedTest
    @ArgumentsSource(V2MemoryRecordsArgumentsProvider.class)
    public void testBaseTimestampToDeleteHorizonConversion(Args args) {
        int partitionLeaderEpoch = 998;
        ByteBuffer buffer = ByteBuffer.allocate(2048);
        MemoryRecordsBuilder builder = MemoryRecords.builder(buffer, args.magic, args.compression, TimestampType.CREATE_TIME,
                0L, RecordBatch.NO_TIMESTAMP, partitionLeaderEpoch);
        builder.append(5L, "0".getBytes(), "0".getBytes());
        builder.append(10L, "1".getBytes(), null);
        builder.append(15L, "2".getBytes(), "2".getBytes());

        ByteBuffer filtered = ByteBuffer.allocate(2048);
        final long deleteHorizon = Integer.MAX_VALUE / 2;
        final RecordFilter recordFilter = new MemoryRecords.RecordFilter(deleteHorizon - 1, 1) {
            @Override
            protected boolean shouldRetainRecord(RecordBatch recordBatch, Record record) {
                return true;
            }

            @Override
            protected BatchRetentionResult checkBatchRetention(RecordBatch batch) {
                return new BatchRetentionResult(BatchRetention.RETAIN_EMPTY, false);
            }
        };
        builder.build().filterTo(new TopicPartition("random", 0), recordFilter, filtered, Integer.MAX_VALUE, BufferSupplier.NO_CACHING);
        filtered.flip();
        MemoryRecords filteredRecords = MemoryRecords.readableRecords(filtered);

        List<MutableRecordBatch> batches = TestUtils.toList(filteredRecords.batches());
        assertEquals(1, batches.size());
        assertEquals(OptionalLong.of(deleteHorizon), batches.get(0).deleteHorizonMs());

        CloseableIterator<Record> recordIterator = batches.get(0).streamingIterator(BufferSupplier.create());
        Record record = recordIterator.next();
        assertEquals(5L, record.timestamp());
        record = recordIterator.next();
        assertEquals(10L, record.timestamp());
        record = recordIterator.next();
        assertEquals(15L, record.timestamp());
        recordIterator.close();
    }
This test is used to see if the base timestamp of the batch has been successfully
converted to a delete horizon for the tombstones / transaction markers of the batch.
It also verifies that the record timestamps remain correct as a delta relative to the delete horizon.
clients/src/test/java/org/apache/kafka/common/record/MemoryRecordsTest.java