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

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.test.util.AbstractTestBase;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.junit.Test;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class FlinkCEPTest extends AbstractTestBase {

    @Test
    public void testFlinkCEP() throws Exception {
        // 创建Kafka消费者
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("test-topic"));

        // 配置Flink CEP环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 编写测试代码
        env.addSource(new SourceFunction<String>() {
            private volatile boolean running = true;

            @Override
            public void run(SourceContext<String> ctx) throws Exception {
                while (running) {
                    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                    for (ConsumerRecord<String, String> record : records) {
                        ctx.collect(record.value());
                    }
                }
            }

            @Override
            public void cancel() {
                running = false;
            }
        })
                .keyBy(value -> value)
                .timeWindow(Time.seconds(10))
                .apply((key, window, input, out) -> {
                    // Flink CEP处理逻辑
                    Pattern<String, ?> pattern = Pattern.<String>begin("start")
                            .where(new SimpleCondition<String>() {
                                @Override
                                public boolean filter(String value) throws Exception {
                                    return value.startsWith("start");
                                }
                            })
                            .followedBy("middle")
                            .where(new IterativeCondition<String>() {
                                @Override
                                public boolean filter(String value, Context<String> ctx) throws Exception {
                                    return value.contains("middle");
                                }
                            })
                            .followedBy("end")
                            .where(new SimpleCondition<String>() {
                                @Override
                                public boolean filter(String value) throws Exception {
                                    return value.endsWith("end");
                                }
                            });

                    pattern.select(input)
                            .forEach((Map<String, List<String>> patternMap) -> {
                                // 处理匹配到的事件模式
                                out.collect(patternMap.get("start").get(0) + " -> " +
                                        patternMap.get("middle").get(0) + " -> " +
                                        patternMap.get("end").get(0));
                            });
                })
                .print();

        // 运行测试
        env.execute();
    }
}