@ParameterizedTest
    @ArgumentsSource(MemoryRecordsArgumentsProvider.class)
    public void testFilterToPreservesPartitionLeaderEpoch(Args args) {
        byte magic = args.magic;
        int partitionLeaderEpoch = 67;

        ByteBuffer buffer = ByteBuffer.allocate(2048);
        MemoryRecordsBuilder builder = MemoryRecords.builder(buffer, magic, args.compression, TimestampType.CREATE_TIME,
                0L, RecordBatch.NO_TIMESTAMP, partitionLeaderEpoch);
        builder.append(10L, null, "a".getBytes());
        builder.append(11L, "1".getBytes(), "b".getBytes());
        builder.append(12L, null, "c".getBytes());

        ByteBuffer filtered = ByteBuffer.allocate(2048);
        builder.build().filterTo(new TopicPartition("foo", 0), new RetainNonNullKeysFilter(), 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());

        MutableRecordBatch firstBatch = batches.get(0);
        if (magic < MAGIC_VALUE_V2) assertEquals(RecordBatch.NO_PARTITION_LEADER_EPOCH, firstBatch.partitionLeaderEpoch());
        else assertEquals(partitionLeaderEpoch, firstBatch.partitionLeaderEpoch());
    }
This test verifies that the checksum returned for various versions matches hardcoded values to catch unintentional
changes to how the checksum is computed.
clients/src/test/java/org/apache/kafka/common/record/MemoryRecordsTest.java