@Test
    public void testBuildLeaderChangeMessage() {
        final int leaderId = 5;
        final int leaderEpoch = 20;
        final int voterId = 6;
        long initialOffset = 983L;

        LeaderChangeMessage leaderChangeMessage = new LeaderChangeMessage()
            .setLeaderId(leaderId)
            .setVoters(Collections.singletonList(
                new Voter().setVoterId(voterId)));
        ByteBuffer buffer = ByteBuffer.allocate(256);
        MemoryRecords records = MemoryRecords.withLeaderChangeMessage(
            initialOffset,
            System.currentTimeMillis(),
            leaderEpoch,
            buffer,
            leaderChangeMessage
        );

        List<MutableRecordBatch> batches = TestUtils.toList(records.batches());
        assertEquals(1, batches.size());

        RecordBatch batch = batches.get(0);
        assertTrue(batch.isControlBatch());
        assertEquals(initialOffset, batch.baseOffset());
        assertEquals(leaderEpoch, batch.partitionLeaderEpoch());
        assertTrue(batch.isValid());

        List<Record> createdRecords = TestUtils.toList(batch);
        assertEquals(1, createdRecords.size());

        Record record = createdRecords.get(0);
        record.ensureValid();
        assertEquals(ControlRecordType.LEADER_CHANGE, ControlRecordType.parse(record.key()));

        LeaderChangeMessage deserializedMessage = ControlRecordUtils.deserializeLeaderChangeMessage(record);
        assertEquals(leaderId, deserializedMessage.leaderId());
        assertEquals(1, deserializedMessage.voters().size());
        assertEquals(voterId, deserializedMessage.voters().get(0).voterId());
    }
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