public Map<TimerKey<?>, ScheduledCallback> removeReadyTimers() {
    final Map<TimerKey<?>, ScheduledCallback> timers = new TreeMap<>(readyTimers);
    // Remove keys on the map directly instead of using key set iterator and remove all
    // on the key set as it results in duplicate firings due to weakly consistent SetView
    for (TimerKey<?> key : timers.keySet()) {
      readyTimers.remove(key);
    }
    return timers;
  }
For run loop to listen to timer firing so it can schedule the callbacks.
samza-core/src/main/java/org/apache/samza/scheduler/EpochTimeScheduler.java