public static <M, WK> TriggerImpl<M, WK> createTriggerImpl(Trigger<M> trigger, Clock clock, TriggerKey<WK> triggerKey) {

    if (trigger == null) {
      throw new IllegalArgumentException("Trigger must not be null");
    }

    if (trigger instanceof CountTrigger) {
      return new CountTriggerImpl<>((CountTrigger<M>) trigger, triggerKey);
    } else if (trigger instanceof RepeatingTrigger) {
      return new RepeatingTriggerImpl<>((RepeatingTrigger<M>) trigger, clock, triggerKey);
    } else if (trigger instanceof AnyTrigger) {
      return new AnyTriggerImpl<>((AnyTrigger<M>) trigger, clock, triggerKey);
    } else if (trigger instanceof TimeSinceLastMessageTrigger) {
      return new TimeSinceLastMessageTriggerImpl<>((TimeSinceLastMessageTrigger<M>) trigger, clock, triggerKey);
    } else if (trigger instanceof TimeTrigger) {
      return new TimeTriggerImpl((TimeTrigger<M>) trigger, clock, triggerKey);
    } else if (trigger instanceof TimeSinceFirstMessageTrigger) {
      return new TimeSinceFirstMessageTriggerImpl<>((TimeSinceFirstMessageTrigger<M>) trigger, clock, triggerKey);
    }

    throw new SamzaException("No implementation class defined for the trigger  " + trigger.getClass().getCanonicalName());
  }
Factory methods for instantiating {@link TriggerImpl}s from individual {@link Trigger}s.
samza-core/src/main/java/org/apache/samza/operators/triggers/TriggerImpls.java