Hands-on Practical: Building a Custom Trigger

Standard triggers provided by the DataStream API cover common use cases like processing time intervals or element counts. However, production pipelines frequently encounter scenarios where window evaluation must depend on complex, domain-specific logic rather than simple clock ticks or counts. For instance, you might need to emit a window result early because a specific "poison pill" pattern was detected, or you might need to implement a dynamic timeout that adjusts based on the current throughput of the stream.

In this practical section, we will architect a hybrid trigger mechanism often required in real-time AI pipelines: a Time-Aware Batching Trigger. This pattern is essential for low-latency inference serving where you want to accumulate a batch of records (vectors) to send to a model endpoint. You cannot wait indefinitely for a batch to fill up, nor can you afford to send single records if throughput allows for efficient batching. This trigger will evaluate the window when either a count threshold is reached or a specific timeout period elapses, whichever happens first.

The Trigger Interface

Implementing a custom trigger requires extending the abstract Trigger class. The Flink runtime invokes specific methods on this class based on different signals. Understanding the lifecycle of these methods is critical for correct state management.

The core methods you must implement are:

1. onElement: Called for every element arriving in the window.
2. onProcessingTime: Called when a registered processing-time timer fires.
3. onEventTime: Called when a registered event-time timer fires.
4. clear: Called when the window is purged to clean up any context state.

Each method returns a TriggerResult that dictates the action the window operator should take.

• CONTINUE: Do nothing.
• FIRE: Evaluate the window function (emit results) but retain the window state.
• PURGE: Clear the window content without evaluating.
• FIRE_AND_PURGE: Evaluate the window function and then clear the content.

Logic Flow for Time-Aware Batching

We define our triggering logic formally. Let $N$ be the maximum batch size and $\Delta t$ be the maximum wait time. For a window starting at $t_0$ containing a set of elements $E$, we trigger if:

$|E| \ge N \quad \lor \quad \text{current\_time} \ge t_0 + \Delta t$

The following diagram outlines the decision logic required within the onElement and timer methods.

Flow control for a hybrid count and time trigger. The system prioritizes filling the batch but ensures latency limits are respected via timers.

Implementing the Trigger

We use Flink's managed state within the trigger to track the element count. Since triggers can fire multiple times or persist across failures, the count must be stored in ReducingState or ValueState, not a local variable.

Below is the implementation logic. We use ValueState to hold the current count and the timestamp of the registered timer. This ensures that we can deduplicate timer registrations.

public class BatchingTrigger<T, W extends Window> extends Trigger<T, W> {
    private final long maxCount;
    private final long timeoutMillis;

    // State descriptors to access Flink managed state
    private final ValueStateDescriptor<Long> countStateDesc = 
        new ValueStateDescriptor<>("count", Long.class);
    private final ValueStateDescriptor<Long> timerStateDesc = 
        new ValueStateDescriptor<>("timer", Long.class);

    public BatchingTrigger(long maxCount, long timeoutMillis) {
        this.maxCount = maxCount;
        this.timeoutMillis = timeoutMillis;
    }

    @Override
    public TriggerResult onElement(T element, long timestamp, W window, TriggerContext ctx) throws Exception {
        ValueState<Long> countState = ctx.getPartitionedState(countStateDesc);
        ValueState<Long> timerState = ctx.getPartitionedState(timerStateDesc);

        long currentCount = countState.value() == null ? 0L : countState.value();
        long newCount = currentCount + 1;
        countState.update(newCount);

        // Condition 1: Count threshold reached
        if (newCount >= maxCount) {
            // Clear any pending timer to avoid double firing
            long timerTimestamp = timerState.value() == null ? 0L : timerState.value();
            if (timerTimestamp > 0) {
                ctx.deleteProcessingTimeTimer(timerTimestamp);
                timerState.clear();
            }
            countState.clear();
            return TriggerResult.FIRE_AND_PURGE;
        }

        // Condition 2: Set timer if not already set
        if (timerState.value() == null) {
            long timer = ctx.getCurrentProcessingTime() + timeoutMillis;
            ctx.registerProcessingTimeTimer(timer);
            timerState.update(timer);
        }

        return TriggerResult.CONTINUE;
    }

    @Override
    public TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception {
        // Timer fired implies timeout reached before count filled
        ctx.getPartitionedState(countStateDesc).clear();
        ctx.getPartitionedState(timerStateDesc).clear();
        return TriggerResult.FIRE_AND_PURGE;
    }

    @Override
    public TriggerResult onEventTime(long time, W window, TriggerContext ctx) {
        return TriggerResult.CONTINUE;
    }

    @Override
    public void clear(W window, TriggerContext ctx) throws Exception {
        ValueState<Long> timerState = ctx.getPartitionedState(timerStateDesc);
        long timerTimestamp = timerState.value() == null ? 0L : timerState.value();
        
        if (timerTimestamp > 0) {
            ctx.deleteProcessingTimeTimer(timerTimestamp);
        }
        
        ctx.getPartitionedState(countStateDesc).clear();
        timerState.clear();
    }
}

Analyzing Trigger Latency

The choice of maxCount and timeoutMillis creates a trade-off between throughput (batch efficiency) and latency. A purely count-based trigger offers optimal throughput but unbounded latency during low traffic. A purely time-based trigger offers bounded latency but potentially small, inefficient batches.

The chart below visualizes the latency profile of this hybrid trigger under varying input rates. Note how the latency is capped at the timeout value even when the arrival rate drops significantly.

Comparison of latency between a fixed count trigger and the hybrid trigger. The hybrid approach caps the maximum latency at 200ms, effectively managing the "long tail" delay seen in the fixed count strategy during low throughput.

State Cleanup and Best Practices

When implementing custom triggers, the clear() method is not optional. Flink calls clear() when a window is purged (e.g., due to allowed lateness expiration) or when the window is explicitly purged by another trigger result. Failure to clean up state results in a resource leak.

In the implementation above, we specifically remove the registered processing time timer. While Flink automatically cleans up window state when a window expires, manually registered timers in the TriggerContext persist unless explicitly deleted. If your stream generates millions of windows, orphaned timers can significantly degrade the performance of the checkpointing mechanism and increase heap usage.

Integration with Window API

To utilize the BatchingTrigger, you attach it to a keyed stream window definition. This replaces the default trigger (usually EventTimeTrigger).

DataStream<Batch> batches = inputStream
    .keyBy(Event::getKey)
    .window(GlobalWindows.create())
    .trigger(new BatchingTrigger<>(10, 200))
    .process(new BatchInferenceFunction());

We typically use GlobalWindows with this pattern because the trigger itself manages the lifecycle of the window (via FIRE_AND_PURGE). The logic does not depend on fixed start and end times but rather on the data arrival patterns. This configuration effectively creates data-driven windows that adapt to the velocity of your stream.