@Override
    public void advanceWatermark(long time) throws Exception {
        currentWatermark = time;

        InternalTimer<K, N> timer;

        while ((timer = eventTimeTimersQueue.peek()) != null
                && timer.getTimestamp() <= time
                && !cancellationContext.isCancelled()) {
            eventTimeTimersQueue.poll();
            final InternalTimer<K, N> timerToTrigger = timer;
            maintainContextAndProcess(
                    timerToTrigger, () -> triggerTarget.onEventTime(timerToTrigger));
            taskIOMetricGroup.getNumFiredTimers().inc();
        }
    }
@Test
    void testEventTimerFireOrder() throws Exception {
        keyContext.setCurrentKey("key-1");
        service.registerEventTimeTimer("event-timer-1", 1L);

        TestTriggerable testTriggerable = new TestTriggerable();
        service.startTimerService(
                IntSerializer.INSTANCE, StringSerializer.INSTANCE, testTriggerable);
        assertThat(testTriggerable.eventTriggerCount).isEqualTo(0);
        // the event timer should be triggered at watermark 1
        service.advanceWatermark(1L);
        assertThat(testTriggerable.eventTriggerCount).isEqualTo(1);

        keyContext.setCurrentKey("key-2");
        service.registerEventTimeTimer("event-timer-2", 2L);
        assertThat(testTriggerable.eventTriggerCount).isEqualTo(1);

        RecordContext<String> recordContext =
                asyncExecutionController.buildContext("record2", "key-2");
        asyncExecutionController.setCurrentContext(recordContext);
        asyncExecutionController.handleRequest(null, StateRequestType.VALUE_GET, null);
        service.advanceWatermark(2L);
        // timer fire is blocked by record2's value_get
        assertThat(testTriggerable.eventTriggerCount).isEqualTo(1);
        // record2 finished, key-2 is released, timer fire can be triggered
        recordContext.release();
        service.advanceWatermark(3L);
        assertThat(testTriggerable.eventTriggerCount).isEqualTo(2);
    }