即使状态存在密钥，ProcessElement发出记录

Question

我有一个保存数据2分钟的状态，即使该密钥有一个州，有时仍然会发出记录。

  @Override
    public void processElement(EngagerEvents value, KeyedProcessFunction<String, EngagerEvents, String>.Context ctx, Collector<String> out) throws Exception {
        if (anonymousIdHasBeenSeen.value() == null) {
            System.out.println("time stamp emitting: " +jsonNode.get("server_timestamp"));
            // key is not available in the state
            anonymousIdHasBeenSeen.update(true);
            System.out.println("TIMER START TIME: " +ctx.timestamp());
            out.collect(value.getEventString());
            ctx.timerService().registerProcessingTimeTimer(ctx.timestamp() + (stateTtl * 1000));
        }
    }

TIMER TRIGGER
-------------

  public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out)
            throws Exception {
        // triggers after ttl has passed
        System.out.println("Call back triggered : time : " +timestamp + " value : " +anonymousIdHasBeenSeen.value());
        if (anonymousIdHasBeenSeen.value()) {
            anonymousIdHasBeenSeen.clear();
        }
    }

我的模拟器代码可生产到Kafka的数据 - IM泵送Anonymid 111，5倍

 public static void main(String[] args) {
        int n = 1; // Number of threads
        for (int i = 0; i < n; i++) {
            Thread thread = new Thread(new ExecutorThread());
            thread.start();
        }


    for (int i = 0; i < 5; i++) {
                ProducerRecord<String, String> record = new ProducerRecord<>("topic", key ,""{"anonymousId": "111", "device": "ios"}"");
                try {
                     producer.send(record);
                     Thread.sleep(500);
                } catch (SerializationException | InterruptedException e) {
                    // may need to do something with it
                }
            }

我用Anonymousid键入。就我而言，只有一个Anonymid 111。回电触发器是60秒，

  DataStream<String> keyedStream = mappedEngagerEventsDataStream.keyBy(EngagerEvents::getAnonymousId)
                .process(new KeyedProcessingWithCallBack(Long.parseLong(60))
                .uid("engager-events-keyed-processing");

我在这里做错了什么？我尝试调试设置一个断点，即使控件不在IF条件内（Sout Insid如果未打印条件），我也会看到发出了特定事件。

我在这里做错了吗？为什么即使我的外出也会发出事件。有人可以指出我在这里做错了什么。