...
3) When a record is matched by RecordEvaluator::isEndOfStream(...), SourceReaderBase stops emitting records from this split and informs SplitFetcherManager to stop reading this split.
4) For SQL users, a new connector option 'scan.record.evaluator.class' is added to provide the custom RecordEvaluator class.
Similar workflow can be made to support this feature for PulsarSource.
...