如何使用Trigger和Evictor测试Flink全局窗口

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

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

        DataStream<Event> events = env.fromElements(
                new Event(1, System.currentTimeMillis()),
                new Event(2, System.currentTimeMillis() + 1000),
                new Event(3, System.currentTimeMillis() + 2000)
        ).assignTimestampsAndWatermarks(WatermarkStrategy.forMonotonousTimestamps());

        events
                .windowAll(GlobalWindows.create())
                .trigger(new CustomTrigger())
                .evictor(new CustomEvictor())
                .reduce(new ReduceFunction<Event>() {
                    @Override
                    public Event reduce(Event value1, Event value2) throws Exception {
                        return new Event(value1.getId() + value2.getId(), value1.getTimestamp());
                    }
                })
                .print();

        env.execute("Global Window Example");
    }

    public static class Event {
        private int id;
        private long timestamp;

        public Event(int id, long timestamp) {
            this.id = id;
            this.timestamp = timestamp;
        }

        public int getId() {
            return id;
        }

        public long getTimestamp() {
            return timestamp;
        }

        @Override
        public String toString() {
            return "Event{" +
                    "id=" + id +
                    ", timestamp=" + timestamp +
                    '}';
        }
    }

    public static class CustomTrigger extends Trigger<Event, TimeWindow> {
        @Override
        public TriggerResult onElement(Event element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
            if (timestamp >= window.maxTimestamp() - 1000) {
                return TriggerResult.FIRE;
            }
            ctx.registerEventTimeTimer(window.maxTimestamp());
            return TriggerResult.CONTINUE;
        }

        @Override
        public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
            return TriggerResult.CONTINUE;
        }

        @Override
        public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
            return TriggerResult.FIRE;
        }

        @Override
        public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
        }
    }

    public static class CustomEvictor implements Evictor<Event, TimeWindow> {
        @Override
        public void evictBefore(Iterable<Event> elements, int size, TimeWindow window, EvictorContext evictorContext) {
            // 移除窗口中时间戳小于当前时间减去2秒的数据
            long cutoff = System.currentTimeMillis() - 2000;
            elements.forEach(event -> {
                if (event.getTimestamp() < cutoff) {
                    elements.remove(event);
                }
            });
        }

        @Override
        public void evictAfter(Iterable<Event> elements, int size, TimeWindow window, EvictorContext evictorContext) {
        }
    }
}