Realtime Compute for Apache Flink:Develop custom triggers

What triggers do

1. Monitor data elements entering a window.

2. Determine whether to trigger the processing based on predefined conditions, such as time and event count.

3. Return one of the following operations:

   • CONTINUE: Continue collecting data.

   • FIRE: Trigger computation and retain states.

   • FIRE_AND_PURGE: Trigger computation and clear states.

   • PURGE: Clear states without triggering computation.

Built-in triggers

Flink provides the following built-in triggers for time- or count-based window operations:

In real-world use cases, we often need to:

• Use more than one trigger condition, such as triggering a window after 5 events or 1 minute.

• Trigger a window based on specific events, such as user logout and order completion.

• Manage a window's lifecycle, like early firing, delayed firing, and multiple firing.

• Avoid default behavior's side effects, such as a window being triggered by late data again.

These complex requirements cannot be satisfied by built-in triggers, making custom triggers necessary.

In a 1-hour tumbling event-time window, the window computation is triggered when the window ends. Additionally, an immediate computation is triggered upon the arrival of a user's fifth event within that window. This early trigger fires at most once per window.

Sample code:

public class CustomCountTrigger extends Trigger<UserEvent, TimeWindow> {

    // Record the number of elements of each key in a window
    private final ValueStateDescriptor<Integer> countStateDesc =
            new ValueStateDescriptor<>("count", Integer.class);
    // Record whether computation has already been triggered
    private final ValueStateDescriptor<Boolean> flagStateDesc =
            new ValueStateDescriptor<>("flag", Boolean.class);

    // Call onElement() when a new element enters the window
    @Override
    public TriggerResult onElement(UserEvent event, long timestamp, TimeWindow window, Trigger.TriggerContext ctx) throws Exception {
        // Get the count state of the current key in the current window
        ValueState<Integer> countState = ctx.getPartitionedState(countStateDesc);
        ValueState<Boolean> flagState = ctx.getPartitionedState(flagStateDesc);
        int count = countState.value() == null ? 0 : countState.value();
        boolean flag = flagState.value() == null ? false : flagState.value();

        // Increment count for each element
        count += 1;
        countState.update(count); // Update state
        
        // When the count reaches 5, trigger window computation immediately
        if (count >= 5 && !flag) {
            flagState.update(true); // Update state to ensure this additional computation is triggered only once
            return TriggerResult.FIRE;
        } else {
            return TriggerResult.CONTINUE;
        }
    }

    @Override
    public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext triggerContext) throws Exception {
        // Skip the processing-time timer in this trigger.
        return TriggerResult.CONTINUE;
    }

    // Call onEventTime() when the registered event-time timer fires (e.g., window end time)
    @Override
    public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
            return TriggerResult.FIRE_AND_PURGE; // Trigger is fired and window is cleared
    }

    @Override
    public void clear(TimeWindow timeWindow, TriggerContext ctx) throws Exception {
        // Clean up window states
        ctx.getPartitionedState(countStateDesc).clear();
        ctx.getPartitionedState(flagStateDesc).clear();
    }
}

Use case 1

Trigger computation more than once before a window ends for alerting.

1. Remove flagStateDesc to enable triggering computation more than once.

2. Alternatively, add a count marker to end a window only after a certain number of trigger firing (such as for alert events).

   ### After removing flagStateDesc, 5 records are ouput.  
   Key: 101, Access count: 5       // count >= 5, triggers early computation and doesn't clear the state.
   Key: 101, Access count: 6       // count >= 5, triggers early computation.
   Key: 101, Access count: 7      // count >= 5, triggers early computation.
   Key: 101, Access count: 8      // count >= 5, triggers early computation.
   Key: 101, Access count: 8      // Trigger computation when the window ends, and clears the state.

3. The trigger only decides when to trigger computation. The actual object information is obtained from .process, where you can specify producing different collect results based on quantity and condition states.

Use case 2

Window closure requires the advancement of watermarks. How can we ensure timely window computation if no data is generated for a long period of time?

• Solution 1: Use processing-time window

  If you don't need event time semantics (namely, the time that events occur is not a concern), you can use processing-time windows:

  .keyBy(keySelector)
  .window(TumblingProcessingTimeWindows.of(Time.seconds(60))) // Use processing time
  .process(new MyProcessWindowFunction())

• Solution 2: Use the withIdleness method

  Flink's WatermarkStrategy provides the withIdleness method to automatically mark a data source as idle after a specified period of inactivity, preventing it from blocking watermark generation.

  // An idle data source will not longer involve in the computation of minimum watermarks, and will not hold back the advancement of watermarks of active data sources.
  WatermarkStrategy
      .<Tuple2<Long, String>>forBoundedOutOfOrderness(Duration.ofSeconds(20))
      .withIdleness(Duration.ofMinutes(1));  //Indicates that if a data source or partition has no events for 1 minute, it is marked as idle

• Solution 3: Use a custom watermark generator

  If you need to use event time semantics, define a custom watermark generator that ensures watermark continues advancing even if no new data arrives:

  • Record the arrival time of the latest event in the onEvent() method.

  • Check the interval between the current time and the last time an event is received in the onPeriodicEmit() method.

  • If the interval exceeds the set threshold, consider the data source as idle, skip generating a watermark or directly generate a specific watermark.

  public class IdleAwareWatermarkGenerator implements WatermarkGenerator<MyEvent> {
      private long lastEventTimestamp = Long.MIN_VALUE;
      private final long maxIdleTimeMs; // Max idle time
  
      public IdleAwareWatermarkGenerator(long maxIdleTimeMs) {
          this.maxIdleTimeMs = maxIdleTimeMs;
      }
  
      @Override
      public void onEvent(MyEvent event, long eventTimestamp, WatermarkOutput output) {
          lastEventTimestamp = Math.max(lastEventTimestamp, eventTimestamp);
      }
  
      @Override
      public void onPeriodicEmit(WatermarkOutput output) {
          long currentTime = System.currentTimeMillis();
          if (lastEventTimestamp == Long.MIN_VALUE || currentTime - lastEventTimestamp > maxIdleTimeMs) {
              // If no events arrive for a long time, don't emit new watermarks
              return;
          }
          output.emitWatermark(new Watermark(lastEventTimestamp));
      }
  }

  WatermarkStrategy<MyEvent> strategy = WatermarkStrategy
      .forGenerator((ctx) -> new IdleAwareWatermarkGenerator(60_000)) // Set maximum idle time to 60 seconds
      .withTimestampAssigner((event, timestamp) -> event.getEventTime());

• Solution 4: Register a timer

  @Override
  public TriggerResult onElement(Event event, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
      ctx.registerEventTimeTimer(window.maxTimestamp()); // Register an event-time timer
      ctx.registerProcessingTimeTimer(window.maxTimestamp() + 1000); // Fault tolerance: ensure a timer is triggered even without events
      return TriggerResult.CONTINUE;
  }
  
  @Override
  public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
      return TriggerResult.FIRE_AND_PURGE; // Forced triggering and cleanup
  }

• Solution 5: Send heartbeat messages from external systems

  Use Kafka or workflows to send regular heartbeat messages downstream to trigger window closure.

Summary