如何使用Kafka消费者为Flink CEP编写Junit测试代码

Question

我们有一个execute()方法，其中我们使用FlinkKafkaConsumer08作为我们的Flink CEP源，然后我们有CEP模式，并且alert再次进入另一个kafka主题。如何为这个execute()方法编写一个junit测试用例？有没有人能为我提供一个样例junit代码？

Pattern.<WebConnectionUseCase>begin("start")
                .where(new SimpleCondition<WebConnectionUseCase>() {
                    public boolean filter(WebConnectionUseCase event) {
                        return ((event.getValues().getPredictedAvailableMemory()
                                - event.getValues().getAvailableMemory()) >= STARTDIFF);
                    }
                }).followedBy("middle").where(new IterativeCondition<WebConnectionUseCase>() {
                    public boolean filter(WebConnectionUseCase value, Context<WebConnectionUseCase> ctx)
                            throws Exception {

                        Iterable<WebConnectionUseCase> middleStops = ctx.getEventsForPattern("middle");
                        List<Double> diffMemoryList = new ArrayList<Double>();
                        List<Double> connectionList = new ArrayList<Double>();
                        middleStops.forEach(item -> diffMemoryList.add(item.getValues().getPredictedAvailableMemory()
                                - item.getValues().getAvailableMemory()));
                        middleStops.forEach(item -> connectionList.add(item.getValues().getConnection()));
                        return checkIncreasingPattern(diffMemoryList) && checkDecreasingPattern(connectionList);
                    }

                    private boolean checkDecreasingPattern(List<Double> list) {
                        //code
                    }

                    private boolean checkIncreasingPattern(List<Double> list) {
                        // code
                    }
                }).times(PATTERNCOUNT).consecutive().next("end").where(new SimpleCondition<WebConnectionUseCase>() {
                    @Override
                    public boolean filter(WebConnectionUseCase event) {
                        return ((event.getValues().getPredictedAvailableMemory()
                                - event.getValues().getAvailableMemory()) >= ENDDIFF);
                    }
                }).within(Time.minutes(TIMEOUTDURATION));

Answer 1

我会将您想要测试的部分封装在一个对象中，您可以连接到特殊的数据源和接收器进行测试，也可以连接到实时数据源/接收器进行生产。

对于测试接收器，您可以使用以下命令：

public static class TestSink<OUT> implements SinkFunction<OUT> {

    // must be static
    public static final List values = new ArrayList<>();

    @Override
    public void invoke(OUT value, Context context) throws Exception {
        values.add(value);
    }
}

然后，您的测试可以将sink.values与预期结果进行比较。

编写进行事件时间处理的测试更容易(与使用处理时间的测试相比)，因为处理时间的结果是不确定的。而且，为了得到确定性的结果，使用并行度为1的测试更容易。

您将找到一些测试here的示例。