@Test
    public void testEmptyBatchRetention() {
        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) {
                        // retain all batches
                        return new BatchRetentionResult(BatchRetention.RETAIN_EMPTY, false);
                    }

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

        // Verify filter result
        assertEquals(0, filterResult.messagesRead());
        assertEquals(records.sizeInBytes(), filterResult.bytesRead());
        assertEquals(baseOffset, filterResult.maxOffset());
        assertEquals(0, filterResult.messagesRetained());
        assertEquals(DefaultRecordBatch.RECORD_BATCH_OVERHEAD, filterResult.bytesRetained());
        assertEquals(timestamp, filterResult.maxTimestamp());
        assertEquals(baseOffset, filterResult.shallowOffsetOfMaxTimestamp());
        assertTrue(filterResult.outputBuffer().position() > 0);

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