Function filter_batch 

pub fn filter_batch<F>(batch: &RecordBatch, keep: F) -> FilteredBatch
where
    F: FnMut(&Record) -> bool,

Drop the records of batch for which keep returns false.

keep runs exactly once per record, in batch order, so a caller may use it to collect per-record facts as a side effect.

The result keeps Kafka’s compacted-log shape: base_offset never moves, each surviving record keeps its offset_delta, and last_offset_delta becomes the highest surviving delta. Every other header field carries over unchanged. That includes max_timestamp, which stays put even when the record that carried the newest timestamp is dropped, because the time index and the retention check read the header and must not see the batch move backwards in time.

A batch that holds no records is FilteredBatch::Unchanged, not FilteredBatch::Empty: nothing was dropped from it. A bare header that an earlier rewrite left behind therefore survives this one intact.