以下是我的flink工作流程：

DataStream<FlinkEvent> events = env.addSource( consumer ).flatMap(...).assignTimestampsAndWatermarks( new EventTsExtractor() );
DataStream<SessionStatEvent> sessionEvents = events.keyBy( 
    new KeySelector<FlinkEvent, Tuple2<String, String> >()
            {
                @Override
                public Tuple2<String, String> getKey( FlinkEvent value ) throws Exception {
                    return(Tuple2.of( value.getF0(), value.getSessionID ) );
                }
            } )
      .window( TumblingEventTimeWindows.of( Time.minutes( 2 ) ) )
      .allowedLateness( Time.seconds( 10 ) )
      .aggregate( new SessionStatAggregator(), new SessionStatProcessor() );
/* ... */
sessionEvents.addSink( esSinkBuilder.build() );

我第一次遇到
java.lang.exception:org.apache.flink.streaming.runtime.tasks.exceptioninchainedoperatorexception:无法将元素转发给下一个运算符
在 flatMap 操作员和任务继续重新启动。我通过相同的键和窗口时间戳观察到许多不同值的重复结果。
问题1：我猜重复是因为下游操作员在作业重新启动后重复使用消息。我说得对吗？我在修正了错误后重新提交了这份工作 ExceptionInChainedOperatorException 问题。我又在第一个时间窗口中观察到了复制品。在那之后，这项工作似乎很顺利（每个键在一个时间窗口中有一个结果）。
问题2：复制品是从哪里来的？