试图实现一个flink作业来读取kafka流并聚合会话，由于某些原因，没有调用getresult（）。我看到createaccumulator（）和add（）被调用，我希望getresult（）也被调用，这样我就可以在目的地接收聚合的消息。

source.keyBy(new KeySelector<GenericRecord, String>() {
                    @Override
                    public String getKey(GenericRecord record) {
                        return record.get("id").toString();
                    }})
                .assignTimestampsAndWatermarks(new AssignerWithPunctuatedWatermarks<GenericRecord>() {
                    private static final long serialVersionUID = -4834111073247835189L;
                    private final long maxTimeLag = 300000L;

                    @Nullable
                    @Override
                    public Watermark checkAndGetNextWatermark(GenericRecord lastElement, long extractedTimestamp) {
                        return new Watermark(extractedTimestamp - maxTimeLag);
                    }

                    @Override
                    public long extractTimestamp(GenericRecord element, long previousElementTimestamp) {
                        long ts = 1000 * (long)element.get(("timestamp"));
                        return (ts);
                    }
                })
                .map(new ReduceAttributesMap())
                .keyBy(new KeySelector<Tuple2<String, String>, String>() {
                    @Override
                    public String getKey(Tuple2<String, String> e) {
                        return e.f0;
                    }
                })
                .window(EventTimeSessionWindows.withGap(Time.minutes(5)))
                .aggregate(new EventAggregation())
                .addSink(...)

有什么问题吗？我是不是误解了什么？谢谢你的帮助！