如何在apache-flink中创建动态规则？

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;

public class DynamicRuleExample {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 模拟数据源
        env.addSource(new SourceFunction<Event>() {
            private volatile boolean isRunning = true;

            @Override
            public void run(SourceContext<Event> ctx) throws Exception {
                Random random = new Random();
                while (isRunning) {
                    ctx.collect(new Event(random.nextInt(100)));
                    Thread.sleep(100);
                }
            }

            @Override
            public void cancel() {
                isRunning = false;
            }
        })
        .keyBy(Event::getValue)
        .timeWindow(TimeWindows.of(Duration.ofSeconds(5)))
        .process(new ProcessWindowFunction<Event, String, Integer, TimeWindow>() {
            @Override
            public void process(Integer key, Context context, Iterable<Event> events, Collector<String> out) throws Exception {
                // 获取当前规则
                int threshold = getRuleFromDatabase(key);

                long count = 0;
                for (Event event : events) {
                    count++;
                }

                if (count > threshold) {
                    out.collect("Threshold exceeded for key: " + key);
                }
            }
        })
        .print();

        env.execute("Dynamic Rule Example");
    }

    private static int getRuleFromDatabase(Integer key) {
        // 模拟从数据库获取规则
        return new Random().nextInt(10);
    }

    public static class Event {
        private int value;

        public Event(int value) {
            this.value = value;
        }

        public int getValue() {
            return value;
        }
    }
}