private static BatchFilterResult filterBatch(RecordBatch batch,
                                                 BufferSupplier decompressionBufferSupplier,
                                                 FilterResult filterResult,
                                                 RecordFilter filter,
                                                 byte batchMagic,
                                                 boolean writeOriginalBatch,
                                                 List<Record> retainedRecords) {
        long maxOffset = -1;
        boolean containsTombstones = false;
        try (final CloseableIterator<Record> iterator = batch.streamingIterator(decompressionBufferSupplier)) {
            while (iterator.hasNext()) {
                Record record = iterator.next();
                filterResult.messagesRead += 1;

                if (filter.shouldRetainRecord(batch, record)) {
                    // Check for log corruption due to KAFKA-4298. If we find it, make sure that we overwrite
                    // the corrupted batch with correct data.
                    if (!record.hasMagic(batchMagic))
                        writeOriginalBatch = false;

                    if (record.offset() > maxOffset)
                        maxOffset = record.offset();

                    retainedRecords.add(record);

                    if (!record.hasValue()) {
                        containsTombstones = true;
                    }
                } else {
                    writeOriginalBatch = false;
                }
            }
            return new BatchFilterResult(writeOriginalBatch, containsTombstones, maxOffset);
        }
    }
Note: This method is also used to convert the first timestamp of the batch (which is usually the timestamp of the first record)
to the delete horizon of the tombstones or txn markers which are present in the batch.
clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java