public void onMessage(M message, TriggerScheduler<WK> context) {
    if (cancellable == null && !shouldFire) {
      final long now = clock.currentTimeMillis();
      long triggerDurationMs = trigger.getDuration().toMillis();
      Long callbackTime = now + triggerDurationMs;
      cancellable =  context.scheduleCallback(() -> {
        LOG.trace("Time since first message trigger fired");
        shouldFire = true;
      }, callbackTime, triggerKey);
    }
  }
Implementation class for a {@link TimeSinceFirstMessageTrigger}
@param <M> the type of the incoming message
samza-core/src/main/java/org/apache/samza/operators/triggers/TimeSinceFirstMessageTriggerImpl.java