@Override
  public void onMessage(M message, TriggerScheduler<WK> context) {
    if (!shouldFire) {
      long currTime = clock.currentTimeMillis();

      if (currTime < callbackTime && cancellable != null) {
        cancellable.cancel();
      }

      callbackTime = currTime + durationMs;
      Runnable runnable = () -> {
        LOG.trace("Time since last message trigger fired");
        shouldFire = true;
      };

      cancellable = context.scheduleCallback(runnable, callbackTime, triggerKey);
    }
  }
Implementation class for a {@link TimeSinceLastMessageTrigger}
@param <M> the type of the incoming message
samza-core/src/main/java/org/apache/samza/operators/triggers/TimeSinceLastMessageTriggerImpl.java