public List<TriggerKey<WK>> runPendingCallbacks() {
    TriggerCallbackState<WK> state;
    List<TriggerKey<WK>> keys = new ArrayList<>();
    long now = clock.currentTimeMillis();

    while ((state = pendingCallbacks.peek()) != null && state.getScheduledTimeMs() <= now) {
      pendingCallbacks.remove();
      state.getCallback().run();
      TriggerKey<WK> key = state.getTriggerKey();
      keys.add(key);
    }
    return keys;
  }
Run all pending callbacks that are ready to be scheduled. A callback is defined as "ready" if it's scheduledTime
is less than or equal to {@link Clock#currentTimeMillis()}

@return the list of {@link TriggerKey}s corresponding to the callbacks that were run.
samza-core/src/main/java/org/apache/samza/operators/impl/TriggerScheduler.java