@ParameterizedTest
    @ArgumentsSource(MemoryRecordsArgumentsProvider.class)
    public void testFilterToEmptyBatchRetention(Args args) {
        byte magic = args.magic;
        for (boolean isTransactional : Arrays.asList(true, false)) {
            ByteBuffer buffer = ByteBuffer.allocate(2048);
            long producerId = 23L;
            short producerEpoch = 5;
            long baseOffset = 3L;
            int baseSequence = 10;
            int partitionLeaderEpoch = 293;
            int numRecords = 2;

            Supplier<MemoryRecordsBuilder> supplier = () -> MemoryRecords.builder(buffer, magic, args.compression, TimestampType.CREATE_TIME,
                    baseOffset, RecordBatch.NO_TIMESTAMP, producerId, producerEpoch, baseSequence, isTransactional,
                    partitionLeaderEpoch);

            if (isTransactional && magic < RecordBatch.MAGIC_VALUE_V2) assertThrows(IllegalArgumentException.class, supplier::get);
            else {
                MemoryRecordsBuilder builder = supplier.get();
                builder.append(11L, "2".getBytes(), "b".getBytes());
                builder.append(12L, "3".getBytes(), "c".getBytes());
                if (magic < MAGIC_VALUE_V2) assertThrows(IllegalArgumentException.class, builder::close);
                else {
                    builder.close();
                    MemoryRecords records = builder.build();
                    ByteBuffer filtered = ByteBuffer.allocate(2048);
                    MemoryRecords.FilterResult filterResult = records.filterTo(new TopicPartition("foo", 0),
                            new MemoryRecords.RecordFilter(0, 0) {
                                @Override
                                protected BatchRetentionResult checkBatchRetention(RecordBatch batch) {
                                    // retain all batches
                                    return new BatchRetentionResult(BatchRetention.RETAIN_EMPTY, false);
                                }

                                @Override
                                protected boolean shouldRetainRecord(RecordBatch recordBatch, Record record) {
                                    // delete the records
                                    return false;
                                }
                            }, filtered, Integer.MAX_VALUE, BufferSupplier.NO_CACHING);

                    // Verify filter result
                    assertEquals(numRecords, filterResult.messagesRead());
                    assertEquals(records.sizeInBytes(), filterResult.bytesRead());
                    assertEquals(baseOffset + 1, filterResult.maxOffset());
                    assertEquals(0, filterResult.messagesRetained());
                    assertEquals(DefaultRecordBatch.RECORD_BATCH_OVERHEAD, filterResult.bytesRetained());
                    assertEquals(12, filterResult.maxTimestamp());
                    assertEquals(baseOffset + 1, filterResult.shallowOffsetOfMaxTimestamp());

                    // Verify filtered records
                    filtered.flip();
                    MemoryRecords filteredRecords = MemoryRecords.readableRecords(filtered);

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

                    MutableRecordBatch batch = batches.get(0);
                    assertEquals(0, batch.countOrNull().intValue());
                    assertEquals(12L, batch.maxTimestamp());
                    assertEquals(TimestampType.CREATE_TIME, batch.timestampType());
                    assertEquals(baseOffset, batch.baseOffset());
                    assertEquals(baseOffset + 1, batch.lastOffset());
                    assertEquals(baseSequence, batch.baseSequence());
                    assertEquals(baseSequence + 1, batch.lastSequence());
                    assertEquals(isTransactional, batch.isTransactional());
                }
            }
        }
    }
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