Flink Filter function Testing-CodePudding

I am trying to find the way to test RichFilter function just like RichMappFunction like present here : RichMapFunction Testing is present here......Searched through internet and find nothing...Please help...

public class Person{
     public String name;
     public Int age;
}

public class OCRKafkaFilter extends RichFilterFunction<Person> {
   
    @Override
    public boolean filter(Person person) {
        return person.age>21;
    }
}

How can we write a test case for this rich filter function for stateful operators ...

CodePudding user response：

I do hope this doc can help you. And here is an example.

CodePudding user response：

You can up a test harness for a filter function doing something like this:

    private OCRKafkaFilter function;
    private OneInputStreamOperatorTestHarness<Person, Person> harness;

    @BeforeEach
    void setUpHarness() throws Exception {
        function = new OCRKafkaFilter();
        StreamFilter<Person> operator = new StreamFilter<>(function);
        harness = new OneInputStreamOperatorTestHarness<>(operator);
        harness.open();
    }