[FLINK-23690] Processing timers can be triggered more efficiently

• Type: Improvement
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Affects Version/s: 1.11.4, 1.14.0, 1.12.5, 1.13.2
• Component/s: Runtime / Task
• Assignee: Unassigned
• Reporter: Jiayi Liao
• Created: 09/Aug/21 12:19
• Updated: 09/Aug/21 12:21

Description

After FLINK-23208, the processing timers are triggered more efficiently but it can still be improved. (The performance can be tested with benchmark)

Currently InternalTimerService.onProcessingTime(long) polls a timer from processingTimeTimersQueue and register a new timer after the polled timer is triggered, which means timers with different timestamps will be registered for multiple times. This can be improved with codes below:

long now = System.currentTimeMillis() - 1
while ((timer = processingTimeTimersQueue.peek()) != null && timer.getTimestamp() <= now) {
    processingTimeTimersQueue.poll();
    keyContext.setCurrentKey(timer.getKey());
    triggerTarget.onProcessingTime(timer);
}

But due to the bug described in FLINK-23689, this change has conflicts with current implementation of TestProcessingTimeService.setCurrentTime(long), which causes a lot of tests to fail(e.g. InternalTimerServiceImplTest). Therefore, before working on this improvement, FLINK-23689 should be fixed firstly.

Issue Links

• is blocked by: FLINK-23689 TestProcessingTimeService.setCurrentTime(long) not delay the firing timer by 1ms delay
• relates to: FLINK-23208 Late processing timers need to wait 1ms at least to be fired