Re: [PR] [FLINK-23208] Add a benchmark for processing timers [flink-benchmarks]

["zhiyisun (via GitHub)" <gi...@apache.org>]

zhiyisun commented on PR #25:
URL: https://github.com/apache/flink-benchmarks/pull/25#issuecomment-2340079177

   # Issue Report: Mismatch in Processing Timers Benchmark
   I encountered an unexpected issue when running the processing timers benchmark on AWS EC2 instances with different CPU configurations. Specifically, I observed that lower-performance CPUs (m6a.2xlarge) outperformed higher-performance ones (m7a.2xlarge).
   
   ```bash
   m7a.2xlarge
   Benchmark                                       Mode  Cnt   Score    Error   Units
   ProcessingTimerBenchmark.fireProcessingTimers  thrpt   30  63.653 ± 10.217  ops/ms
   ```
   
   ```bash
   m6a.2xlarge
   Benchmark                                       Mode  Cnt    Score    Error   Units
   ProcessingTimerBenchmark.fireProcessingTimers  thrpt   30  108.015 ± 22.383  ops/ms
   ```
   
   # Investigation and Findings:
   Further investigation revealed a discrepancy in the timer registration mechanism used by the benchmark. In ProcessingTimerKeyedProcessFunction::processElement, timers are registered in reverse chronological order. This leads to previously added timers being canceled upon new registrations.
   
   ```java
               for (int i = 0; i < timersPerRecord; i++) {
                   context.timerService().registerProcessingTimeTimer(currTimestamp - i - 1);
               }
   ```
   
   When the benchmark registers all timers with timestamps older than the current time, it causes these timers to be scheduled immediately. The nextTimer.cancel(false) in InternalTimerServiceImpl::registerProcessingTimeTimer doesn't cancel the timer for each iteration successfully. 
   
   ```java
       public void registerProcessingTimeTimer(N namespace, long time) {
           InternalTimer<K, N> oldHead = processingTimeTimersQueue.peek();
           if (processingTimeTimersQueue.add(
                   new TimerHeapInternalTimer<>(time, (K) keyContext.getCurrentKey(), namespace))) {
               long nextTriggerTime = oldHead != null ? oldHead.getTimestamp() : Long.MAX_VALUE;
               // check if we need to re-schedule our timer to earlier
               if (time < nextTriggerTime) {
                   if (nextTimer != null) {
                       nextTimer.cancel(false);
                   }
                   nextTimer = processingTimeService.registerTimer(time, this::onProcessingTime);
               }
           }
       }
   ```
   
   Even if setRemoveOnCancelPolicy is set to true and nextTimer.cancel(true) can't make sure the timer is canceled successfully. I created a simple [sample code](https://github.com/zhiyisun/TaskSchedulingTest) to show this issue. 
   
   Furthermore, InternalTimerServiceImpl::onProcessingTime registers more timers when those canceled timers are called, leading to an accumulation of errors.
   
   ```java
       void onProcessingTime(long time) throws Exception {
           // null out the timer in case the Triggerable calls registerProcessingTimeTimer()
           // inside the callback.
           nextTimer = null;
   
           InternalTimer<K, N> timer;
   
           while ((timer = processingTimeTimersQueue.peek()) != null
                   && timer.getTimestamp() <= time
                   && !cancellationContext.isCancelled()) {
               keyContext.setCurrentKey(timer.getKey());
               processingTimeTimersQueue.poll();
               triggerTarget.onProcessingTime(timer);
               taskIOMetricGroup.getNumFiredTimers().inc();
           }
   
           if (timer != null && nextTimer == null) {
               nextTimer =
                       processingTimeService.registerTimer(
                               timer.getTimestamp(), this::onProcessingTime);
           }
       }
   ```
   
   # Conclusion and Recommendations:
   While the timer mechanism in Flink is sound, this specific benchmark does not accurately reflect real-world usage. As a result, the scores generated by this benchmark are misleading.
   
   To address this issue, I recommend either:
   1. Disabling the benchmark by default: This would prevent users from relying on inaccurate results.
   2. Modifying the code to register timers in chronological order: This way won't be able to leverage the advantage of the batch timer callback process (while loop) in InternalTimerServiceImpl::onProcessingTime(). The timer will be registered one at a time. But this can reflect the system performance.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org