@Test
    public void testEmptyBatchDeletion() {
        for (final BatchRetention deleteRetention : Arrays.asList(BatchRetention.DELETE, BatchRetention.DELETE_EMPTY)) {
            ByteBuffer buffer = ByteBuffer.allocate(DefaultRecordBatch.RECORD_BATCH_OVERHEAD);
            long producerId = 23L;
            short producerEpoch = 5;
            long baseOffset = 3L;
            int baseSequence = 10;
            int partitionLeaderEpoch = 293;
            long timestamp = System.currentTimeMillis();

            DefaultRecordBatch.writeEmptyHeader(buffer, RecordBatch.MAGIC_VALUE_V2, producerId, producerEpoch,
                    baseSequence, baseOffset, baseOffset, partitionLeaderEpoch, TimestampType.CREATE_TIME,
                    timestamp, false, false);
            buffer.flip();

            ByteBuffer filtered = ByteBuffer.allocate(2048);
            MemoryRecords records = MemoryRecords.readableRecords(buffer);
            MemoryRecords.FilterResult filterResult = records.filterTo(new TopicPartition("foo", 0),
                    new MemoryRecords.RecordFilter(0, 0) {
                        @Override
                        protected BatchRetentionResult checkBatchRetention(RecordBatch batch) {
                            return new BatchRetentionResult(deleteRetention, false);
                        }

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

            // Verify filter result
            assertEquals(0, filterResult.outputBuffer().position());

            // Verify filtered records
            filtered.flip();
            MemoryRecords filteredRecords = MemoryRecords.readableRecords(filtered);
            assertEquals(0, filteredRecords.sizeInBytes());
        }
    }
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