public <K> void setTimer(K key, long timestamp, ScheduledCallback<K> callback) {
    if (scheduledFutures.containsKey(key)) {
      LOG.warn("Registering duplicate callback for key: {}. Attempting to cancel the previous callback", key);
      ScheduledFuture<?> scheduledFuture = scheduledFutures.get(key);

      /*
       * We can have a race between the time we check for the presence of the key and the time we attempt to cancel;
       * Hence we check for non-null criteria to ensure the executor hasn't kicked off the callback for the key which
       * removes the future from the map before invoking onTimer.
       *  1. In the event that callback is running then we will not attempt to interrupt the action and
       *     cancel will return as unsuccessful.
       *  2. In case of the callback successfully executed, we want to allow duplicate registration to keep the
       *     behavior consistent with the scenario where the callback is already executed or in progress even before
       *     we entered this condition.
       */
      if (scheduledFuture != null
          && !scheduledFuture.cancel(false)
          && !scheduledFuture.isDone()) {
        LOG.warn("Failed to cancel the previous callback successfully. Ignoring the current request to register new callback");
        return;
      }
    }

    final long delay = timestamp - System.currentTimeMillis();
    final ScheduledFuture<?> scheduledFuture = executor.schedule(() -> {
      scheduledFutures.remove(key);
      readyTimers.put(TimerKey.of(key, timestamp), callback);

      if (timerListener != null) {
        timerListener.onTimer();
      }
    }, delay > 0 ? delay : 0, TimeUnit.MILLISECONDS);
    scheduledFutures.put(key, scheduledFuture);
  }
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