我正在阅读的源代码 CepOperator 来自flink cep，对以下代码段有疑问：

public void processElement(StreamRecord<IN> element) throws Exception {
        if (isProcessingTime) {
            ... ...
        } else {

            long timestamp = element.getTimestamp();
            IN value = element.getValue();

            if (timestamp > lastWatermark) {

                saveRegisterWatermarkTimer();

                bufferEvent(value, timestamp);

            } else if (lateDataOutputTag != null) {
                output.collect(lateDataOutputTag, element);
            } else {
                numLateRecordsDropped.inc();
            }
        }
    }

我不明白为什么每次收到新元素， saveRegisterWatermarkTimer() 接到电话了？以下是源代码：

private void saveRegisterWatermarkTimer() {
    long currentWatermark = timerService.currentWatermark();
    // protect against overflow
    if (currentWatermark + 1 > currentWatermark) {
        timerService.registerEventTimeTimer(VoidNamespace.INSTANCE, currentWatermark + 1);
    }
}

它几乎总是注册一个新的事件时间计时器。它不会产生太多的计时器吗？
谢谢你的解释。