DEV Community: Giannis Polyzos

A Deep Dive Into Apache Flink Timers

Giannis Polyzos — Tue, 04 Mar 2025 09:08:52 +0000

Timers in Apache Flink are scheduled callbacks that can be set to trigger at a specific time in the future.

Flink provides two main notions of time for stream processing: processing time and event time. Timers can be registered for either of these time domains:

⌛Processing-time timers: These timers use the system’s real clock (wall-clock time). A processing-time timer triggers when the machine’s current time reaches the specified timestamp. This is essentially "now" as measured by the processing node. Processing time is straightforward and low-latency (no need for watermark coordination), but it can be less deterministic since it depends on the clock of the machine processing the data.
⏳Event-time timers: These timers use the timestamps of events (the time when an event occurred, as embedded in the event). Flink’s event-time processing is driven by watermarks, special messages that flow with the data to indicate progress in event time. An event-time timer triggers when the operator’s watermark passes the timer’s timestamp. This means all events with a timestamp less than or equal to that time have been processed. Event-time timers allow correct handling of out-of-order events and lateness, providing deterministic results based on event timestamps, regardless of processing delays.

Key differences

👉 Trigger mechanism: A processing-time timer fires based on wall-clock time, whereas an event-time timer fires based on the advancement of watermarks (which in turn reflect event timestamps).

👉 Use cases: Event-time timers are ideal for time-dependent computations that need to respect event timestamps (for example, window aggregations, session gap detection, or complex event patterns). Processing-time timers are useful for tasks like emitting periodic updates, timing out idle sessions, or heartbeat mechanisms that rely on real-time passage.

👉 Determinism: Event-time processing yields consistent results under replay or different speeds (since event order and watermarks dictate the timing), whereas processing-time results can vary if the job is paused or slowed (because the wall clock keeps ticking).

Flink also supports ingestion time (the time events enter Flink) as a middle ground, but in modern Flink APIs, we typically choose between event time and processing time explicitly.

Stateful Processing and Timers in Flink

Most Flink streaming applications are stateful,i.e they remember information across events. Flink’s ProcessFunction gives you access to state and timers, making it a powerful “Swiss army knife” for custom streaming logic.

When using a ProcessFunction -- KeyedProcessFunction for example, Flink provides:

✅ Keyed state: You can declare state (e.g., ValueState, ListState) that is scoped to each key in a keyed stream. Only events with the same key share that state. The state for each key is stored by Flink in the state backend and is checkpointed for fault tolerance. This means your state is persisted and can be recovered on failures. State and timers are kept locally in each task and periodically snapshotted to durable storage for fault tolerance.

✅ TimerService: The context in a ProcessFunctionprovides a TimerService for registering timers. You can schedule a callback for a future time in event time or processing time (in milliseconds). For example, ctx.timerService().registerEventTimeTimer(t) will schedule an onTimer() call for that key when the event-time watermark reaches t, and registerProcessingTimeTimer(t) will trigger when the processing time clock reaches t. The onTimer(...) callback is where you define what should happen when the timer fires.

✅ Timer callbacks: When a timer fires, Flink invokes the onTimer() method in the ProcessFunction. Within onTimer, you can determine the time domain (event or processing) via ctx.timeDomain(), and emit results or update state.

Note: The onTimer callback, like processElement, is executed with the keyed context, meaning you can read or update the keyed state as needed for that key. Flink ensures that calls to processElement and onTimer are thread-safe and serialized per key, so you don’t need extra locking when accessing state.

✅ State and timer persistence: Both state and timers are part of Flink’s managed state and are included in snapshots. Timers are checkpointed along with state. If your job restarts from a checkpoint or savepoint, any timers that were set will be restored. Notably, if a timer was supposed to fire while the job was down (e.g., the checkpoint was taken before the timer timestamp and on restore the current time or watermark is already past that point), Flink will fire the timer immediately upon restore. This guarantees no timer is “missed” across failures. Timers are also automatically deduplicated by Flink: for each key and exact timestamp, only one timer exists – if you register another timer for the same timestamp (and same key), it will not create a duplicate; the callback will only execute once. You can also cancel timers by their timestamp if needed (we’ll see an example of that).

Now that we’ve covered the basics of timers and state, let’s apply them in a concrete example.

Example Use Case: Monitoring Sensor Readings

Let's assume we have a stream of sensor readings from IoT sensors. Each reading has a sensor ID, a timestamp (when the reading was produced), and a temperature value. We want to monitor these readings for two purposes:

Periodic aggregation (event-time): Continuously compute the average temperature of each sensor over 1-minute and output the result. Yes, we could do this with just a time window 🙃, but since the goal is to demonstrate timers here we will do this using an event-time timer (to trigger at the end of each window). This simulates a custom windowing logic using timers and state.
Inactivity alert (processing-time): Detect if a sensor stops sending data for a certain amount of time (e.g., 10 seconds) in real processing time, and raise an alert if so. This will be done with processing-time timers, which naturally fit the real-time notion of “no event received for X seconds”.

Why use different timers for these? The aggregation is aligned with event time because we want the window to consider the event timestamps and handle out-of-order events correctly based on watermarks. The inactivity alert is about a real-time gap in data arrival, which is easier to measure with processing time, since if no events come, event time progress might stall, but processing time will still tick away. This should help understand both kinds of timers.

The full source code can be found here and if you are looking for production options you can check the Ververica Cloud for free.

The core of our job lives in the KeyedProcessFunction that uses state and timers to realize our two goals: 1. average computation and 2. inactivity alert.

For each sensor (key), we need to keep track of:

The running sum of temperatures and count of readings in the current window (to compute the average at the window end).
The end timestamp of the current window, so we know when to trigger output, and to detect when a new event falls into the next window.
The timestamp of the currently scheduled inactivity timer, so we can cancel it if a new event arrives in time.

We will use Flink’s ValueState to store these values.


    @Override
    public void processElement(SensorReading sensorReading, 
                               KeyedProcessFunction<String, SensorReading, String>.Context context,
                               Collector<String> collector) throws Exception {
        // Extract the event timestamp of the current reading
        long eventTime = sensorReading.getTimestamp();
        long currentWindowEnd = windowEndState.value();

        // 1) Event-Time Timer Logic for windowing (1-minute tumbling windows per sensor)
        if (currentWindowEnd == null) {
            // This is the first event for this key or the first event after a window reset
            long windowStart = eventTime - (eventTime % WINDOW_DURATION);
            long windowEnd = windowStart + WINDOW_DURATION;
            // register an event-time timer for end of the window
            context.timerService().registerEventTimeTimer(windowEnd);

            windowEndState.update(windowEnd);
            sumState.update(sensorReading.getTemperature());
            countState.update(1L);
        } else if (eventTime < currentWindowEnd) {
            // Still within the current window
            sumState.update(sumState.value() + sensorReading.getTemperature());
            countState.update(countState.value() + 1);
        } else {
            // The new event belongs to a next window (current window has ended)
            // Emit the result for the current window before resetting
            double sum = sumState.value();
            long count = countState.value();
            double avg = sum / count;
            collector.collect(
                    "Average temperature for sensor " + sensorReading.getSensorId() 
                            + " for window ending at " + currentWindowEnd + " = " + avg
            );
            // Clear the old window state and cancel the old timer
            context.timerService().deleteEventTimeTimer(currentWindowEnd);

            sumState.clear();
            countState.clear();
            windowEndState.clear();

            // Start a new window for the incoming event
            long windowStart = eventTime - (eventTime % WINDOW_DURATION);
            long windowEnd = windowStart + WINDOW_DURATION;

            context.timerService().registerEventTimeTimer(windowEnd);
            windowEndState.update(windowEnd);
            sumState.update(sensorReading.getTemperature());
            countState.update(1L);
        }
        // 2) Processing-Time Timer Logic for inactivity alert (10s of no events)
        //      Every time we get an event, schedule a processing-time timer X ms in the future.
        //      If a new event comes before that, cancel the previous timer and schedule a new one.
        long prevTimerTimestamp = inactivityTimerState.value();
        if (prevTimerTimestamp != null) {
            // Remove the old scheduled timer because we got a new event
            context.timerService().deleteProcessingTimeTimer(prevTimerTimestamp);
        }
        // Register a new processing-time timer for now + threshold
        long newTimerTimestamp = context.timerService().currentProcessingTime() + INACTIVITY_THRESHOLD;
        context.timerService().registerProcessingTimeTimer(newTimerTimestamp);
        // Store the new timer's timestamp in state
        inactivityTimerState.update(newTimerTimestamp);
    }

Let’s break down what’s happening here:

If this is the first event for the sensor (i.e., currentWindowEnd == null because we haven’t seen this key before or we cleared after a window), we determine the current window’s start and end. We align the window to minute boundaries using the timestamp modulo the window size. For example, if an event arrives at 12:34:45.123 (ms since epoch), and our window is 60,000 ms (1 minute), we align to 12:34:00.000 as start and 12:35:00.000 as end. We then register an event-time timer for the window end. This means “when the watermark reaches windowEnd call onTimer for this key.” We initialize the sum and count state with this first event’s values.
If another event comes and its timestamp is still before the current window’s end (eventTime < currentWindowEnd), it’s in the same window. We update the running sum and count in state. (We do not need to register a new timer because one is already set for the window end.)
If an event arrives that is at or beyond the current window’s end (eventTime >= currentWindowEnd), that indicates the previous window should be finished. This can happen if events come slightly out of order or if there was a gap. In this case, we manually close out the current window: calculate the average from sumState and countState, emit it and clear the states. We also cancel the event-time timer for the old window since we’re handling it early – this prevents the onTimer() from later firing for a window we’ve already emitted. Then we start a new window: we determine the new window boundaries based on the current event’s timestamp, register a new event-time timer for the new window’s end, and set the state. In effect, we are handling the window rollover immediately when we see an event from the next window. This logic ensures that an event exactly on a window boundary or slightly late will still be accounted correctly in the next window. In a perfect scenario with in-order data, this branch might not trigger often, but it’s important for correctness with out-of-order events.

In the code, we use a ValueState<Long> (inactivityTimerState) to remember the timestamp of the last scheduled processing-time timer for the key. On each event:

We retrieve the previous timer timestamp (if any). If it exists, we cancel that timer. If the timer had already fired, the state would likely have been cleared; if it hasn’t fired yet, now it won’t because we canceled it.
We then register a new processing-time timer for currentProcessingTime + 10000 ms (10 seconds in the future).
We store this new timer’s timestamp in the state (so that a subsequent event can cancel it if needed).

The actual emission of the results happens in onTimer() method.

  @Override
    public void onTimer(long timestamp, KeyedProcessFunction<String, SensorReading, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
        // Check the domain of the timer (event-time or processing-time)
        if (ctx.timeDomain() == TimeDomain.EVENT_TIME) {
            // Event-time timer fired (window end reached)
            double sum = sumState.value() != null ? sumState.value() : 0.0;
            long count = countState.value() != null ? countState.value() : 0L;
            if (count > 0) {
                double avg = sum / count;
                out.collect("Average temperature for sensor " + ctx.getCurrentKey() +
                        " for window ending at " + timestamp + " = " + avg);
            }
            // Clear window state after emitting result
            sumState.clear();
            countState.clear();
            windowEndState.clear();
        } else if (ctx.timeDomain() == TimeDomain.PROCESSING_TIME) {
            // Processing-time timer fired (inactivity threshold passed)
            out.collect("ALERT: Sensor " + ctx.getCurrentKey() + " has been inactive for " + (INACTIVITY_THRESHOLD / 1000) + " seconds");
            // Clear the inactivity timer state (no active timer now for this key)
            inactivityTimerState.clear();
        }
    }

Running and Testing the Flink Job

If you run the application you should see the following output:

Average temperature for sensor sensor_2 for window ending at 60000 = 30.0
Average temperature for sensor sensor_1 for window ending at 60000 = 22.625
Average temperature for sensor sensor_2 for window ending at 120000 = 32.0
ALERT: Sensor sensor_1 has been inactive for 10 seconds
ALERT: Sensor sensor_2 has been inactive for 10 seconds

Note: In order to see the above output though, some "cheating" is required 😊 Since we have a bounded source, this means that the program will exit as soon as its consumed. So we need to add some delay to make sure 10 seconds of processing time are passed, see here

For testing, you can adjust the input events or durations to see different scenarios, such as a sensor not sending any data (to quickly trigger the inactivity alert), or events coming slightly out-of-order. The WatermarkStrategy with 5 seconds out-of-orderness means the window result will wait until 5 seconds after the window to emit (to allow late arrivals) – you can lower this for faster results if needed.

Conclusion

Apache Flink’s timers are really useful for building responsive, stateful streaming applications. They allow you to implement custom time-based logic that goes beyond the built-in window and time-based operators – for example, detecting complex patterns, implementing custom windowing, or monitoring inactivity as we did.

Some key takeaways and best practices:

Decide between event time and processing time based on your use case. Use event time for correctness w.r.t event timestamps and handling of out-of-order events (with watermarks), and processing time for real-time driven actions where slight nondeterminism is acceptable.
Key your stream before using timers, as timers are per key. This also helps isolate state per entity (like sensor) for scalability.
Clean up state when it’s no longer needed (we cleared our window state after use, and managed the cancelation of timers). This prevents state from growing indefinitely for inactive keys.
Remember that timers are part of state – they will survive failures and resume, firing if their time is due.
Monitor event-time progress (watermarks) when using event-time timers. If watermarks don’t advance (e.g., if sources become idle and you haven’t marked them as idle), event-time timers won’t fire.

By leveraging timers and managed state, you can tackle a wide array of streaming scenarios that require remembering information and acting on time-based conditions. Apache Flink handles the heavy lifting of making this reliable and scalable under the hood.

You made it and reached the end. I hope you enjoyed this 👋 and happy streaming 🌊

Understanding Custom Triggers In Apache Flink

Giannis Polyzos — Fri, 28 Feb 2025 08:22:24 +0000

Apache Flink’s window triggers determine when a window’s computation is emitted (and optionally cleared) during stream processing.

Every window in Flink has an associated trigger that evaluates each incoming element and any relevant timer events to decide whether to FIRE (emit results), PURGE (drop window contents), or both.

In other words, the trigger monitors window conditions and tells Flink when to produce the aggregated output of that window. By default, Flink provides built-in triggers for common scenarios, but they have limitations and this is where custom triggers become useful.

Build-in Triggers

Flink comes with a few built-in triggers for standard behaviors:

EventTimeTrigger: fires when the event-time watermark passes the end of the window (i.e. window closes in event time). This is the default for event-time windows and ensures results are emitted once the window’s end timestamp is reached according to the stream’s watermarks.
ProcessingTimeTrigger: fires when the processing-time clock reaches the end of the window. This is the default for processing-time windows.
CountTrigger: fires once the number of elements in the window reaches a specified count threshold
PurgingTrigger: a wrapper that turns any other trigger into a purging trigger, meaning it will clear the window’s content whenever it fires.

These built-in triggers cover basic cases -- time-driven or count-driven firing, but they operate independently. If you override a window’s trigger with one of these, you replace the default trigger, not supplement it. For example, if you apply a CountTrigger to an event-time window, the window will only fire based on count and ignore the event-time watermark. This means built-in triggers can’t be combined out-of-the-box – you can’t directly get a window that fires on either a count or a time condition using only the provided triggers.

In practice, complex scenarios often require custom triggers. Custom triggers let you define arbitrary conditions (or combinations of conditions) for firing windows, such as “fire on count or time, whichever comes first” or “fire when a special event occurs.”

By writing a custom trigger, you can overcome the limitations of the built-ins and implement custom early results, session-specific logic, late firing policies, and more.

Creating Custom Triggers

To implement a custom trigger in Flink (DataStream API), you create a class that extends the abstract Trigger<T, W> class, where T is the type of elements in the window and W is the window type (e.g. TimeWindow).

In your subclass, you override several methods that Flink uses to drive the trigger’s behavior. You can find a full list and description with those methods here.

A TriggerContext is provided to these methods, which offers utilities to register and delete timers (see more here) and to access partitioned state tied to the window. You can use Flink's state within triggers to track information like element counts or flags. For instance, if implementing a count-based trigger, you might keep a ValueState<Integer> for the count of elements seen so far in the window.

Note: Triggers must be serializable as they are part of the job graph sent to workers. Also, remember that returning FIRE does not clear the window state by default – the window remains and may accumulate more data, potentially firing again later. If you want the window to be cleared when it fires (emitting results only once), you should return FIRE_AND_PURGE or use a PurgingTrigger wrapper.

We’ll see examples of both behaviors below, so let's walk through 🚶two real-world examples using custom triggers in Flink.

The full source code can be found here and if you are looking for production options you can check the Ververica Cloud for free.

Example 1: Tumbling Window and Count-Based Early Firing

Scenario: Imagine an e-commerce application tracking user events – e.g. page views, add-to-cart, checkout initiations – as a stream. We want to compute some analytics per user in a tumbling time window (say, 1 hour windows per user).

However, if a user is very active, we don’t want to wait until the end of the hour to get intermediate results. We decide that for each user’s window, the result should be emitted early after 5 events (we will use a small threshold for demonstration purposes) have been received in that window, rather than waiting the full hour.

We will implement a custom trigger that tracks the count of elements in the window. It will fire when the count reaches 5, and also ensure the window will fire at the end of the time window.

The important part is within the CustomCountTrigger class:

 @Override
    public TriggerResult onElement(UserEvent userEvent, 
                                   long timestamp,
                                   TimeWindow timeWindow,
                                   TriggerContext triggerContext) throws Exception {
        // Get or initialize the current count
        ValueState<Integer> countState = triggerContext.getPartitionedState(countStateDesc);
        Integer count = countState.value();
        if (count == null) {
            count = 0;
        }
        // Increment count for every element
        count += 1;
        countState.update(count);

        // If this is the first element, register an event-time timer for end-of-window
        // (Timers are set at window end timestamp, so when watermark passes window.end, onEventTime will fire)
        if (count == 1) {
            long windowEnd = timeWindow.getEnd();  // end timestamp of this TimeWindow
            triggerContext.registerEventTimeTimer(windowEnd);
        }

        // Check if we've reached 5 events in this window
        if (count >= 5) {
            // Fire (emit the window) *now*, but do NOT purge (we return FIRE, not FIRE_AND_PURGE).
            // This means the window contents remain, and the window will possibly fire again at end-of-window.
            return TriggerResult.FIRE;
        } else {
            // Not yet reached 5, so continue accumulating
            return TriggerResult.CONTINUE;
        }    }

    @Override
    public TriggerResult onProcessingTime(long timestamp, TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {
        // We don't use processing-time timers in this trigger.
        return TriggerResult.CONTINUE;
    }

    @Override
    public TriggerResult onEventTime(long timestamp, TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {
        // This is called when the event-time timer for the window fires (i.e., watermark reached window end).
        if (timestamp == timeWindow.getEnd()) {
            // Window end reached, so fire the window result.
            // We return FIRE_AND_PURGE to emit the result and clear the window state.
            return TriggerResult.FIRE_AND_PURGE;
        }
        // If it's not the window-end timer (e.g., some other timer), we ignore.
        return TriggerResult.CONTINUE;        }

    @Override
    public void clear(TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {
        // Clean up the count state when the window is purged/closed.
        ValueState<Integer> countState = triggerContext.getPartitionedState(countStateDesc);
        countState.clear();
        // We don't need to manually delete the event-time timer for window end – Flink does that when window closes.
    }

Let's break this down. On each element:

It registers an event-time timer for the end-of-window when the first element arrives (to ensure the window eventually fires at the end if the count trigger didn’t already fire) – similar to what Flink’s default EventTimeTrigger does internally.
If count reaches 5, it returns TriggerResult.FIRE. We choose FIRE (not FIRE_AND_PURGE) so that the window’s contents are not thrown away; this means more elements can still accumulate for the final output. We could also decide to reset the count state to 0 here if we wanted to allow another early firing every 5 events, but in this example we only fire once early at 5 and then rely on the final timer.
In onEventTime(), when the watermark hits the window end timestamp, we return FIRE_AND_PURGE. This emits the final window result and clears out the window state (the clear() method will be called to drop the count state). We use FIRE_AND_PURGE here because once the window end is reached, the window should close and not hold resources.
The clear() method ensures we remove the count from state to avoid leaks. Flink will call clear() after the window is purged (either via our final fire or if the window is disposed for any reason).

This custom trigger demonstrates combining a count condition with an event-time window trigger. Without a custom trigger, achieving “5 events or end-of-window” would not be possible with Flink’s built-ins alone

If you run the above example you should see the following output:

User user_1 had 5 events in window [2025-02-01 00:00:00.0,2025-02-01 01:00:00.0)
User user_1 had 6 events in window [2025-02-01 00:00:00.0,2025-02-01 01:00:00.0)
User user_2 had 5 events in window [2025-02-01 00:00:00.0,2025-02-01 01:00:00.0)

Notice here that the window gets triggered for user_1 when it gets 5 events and also when the window ends.

Example 2: Session Inactivity With Event Signals

Scenario: Now let’s consider a session window example. In many applications, events are grouped by sessions (a period of user activity separated by inactivity). Flink’s session windows can close a window after a period of inactivity. However, sometimes a session might also end due to a specific event –for example, a user explicitly logs out or completes a checkout.

In an e-commerce or login-based app, you may define a session to end either after 30 minutes of inactivity or when a “logout” or “checkout complete” event occurs.

Flink’s built-in session windows don’t know about the semantics of a logout event, they’d only close based on the timeout. We can create a custom trigger to handle both: fire the window when a special session-end event is seen, or if the session times out with no activity.

The important part is within the SessionEndTrigger class:

 @Override
    public TriggerResult onElement(UserEvent userEvent, long timestamp, TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {
        // Register an event-time timer for the end-of-window (session timeout)
        triggerContext.registerEventTimeTimer(timeWindow.getEnd());
        // If this event is a session terminating event, fire and purge the window
        if ("LOGOUT".equals(userEvent.getEventType()) || "CHECKOUT_COMPLETE".equals(userEvent.getEventType())) {
            return TriggerResult.FIRE_AND_PURGE;
        }
        // Otherwise, continue
        return TriggerResult.CONTINUE;    }

    @Override
    public TriggerResult onProcessingTime(long timestamp, TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {
        // not used in this trigger
        return TriggerResult.CONTINUE;    }

    @Override
    public TriggerResult onEventTime(long timestamp, TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {
        if (timestamp == timeWindow.getEnd()) {
            // Session inactivity timeout reached, fire and purge the window
            return TriggerResult.FIRE_AND_PURGE;
        }
        return TriggerResult.CONTINUE;
    }

    @Override
    public void onMerge(TimeWindow window, OnMergeContext ctx) throws Exception {
        // If windows merge, register a new timer for the new window end and let the old timers lapse.
        // (Flink will call onEventTime for the exact timestamps that occur; by re-registering the new end we ensure final firing.)
        ctx.registerEventTimeTimer(window.getEnd());
    }

    @Override
    public boolean canMerge() {
        return true;
    }

    @Override
    public void clear(TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {
        triggerContext.deleteEventTimeTimer(timeWindow.maxTimestamp());
    }

Let's break this down. On each element:

We register a timer, which ensures an event-time timer is set for the current end-of-window. (If the session window extends or merges, window.getEnd() will be updated accordingly on each trigger invocation.)
We then check the event’s type. If it is a designated end-of-session event (in this case, "LOGOUT" or "CHECKOUT_COMPLETE"), we immediately return FIRE_AND_PURGE. This emits the window’s results and clears the window state right away, effectively closing the session as soon as that event is encountered.
If it’s not an end signal, we return CONTINUE to keep collecting events.
In onEventTime(), we check if the timer is at the window’s end timestamp. If yes, that means the watermark has reached the session’s end (i.e., no new events for 15 minutes), so we return FIRE_AND_PURGE to close out the session. (If the timer firing is for some other timestamp, we ignore it with CONTINUE.)
We don’t use onProcessingTime() here, so it just continues.
The clear() method doesn’t have to do much because we didn’t use manual state (timers will be cleaned up by Flink after purging). If we had used any ValueState in the trigger, we’d clear it here.

If you run the code above you should see the following output:

Session for user user_1 [2025-02-01 00:20:00.0,2025-02-01 00:35:20.0) -> Events: [login, click, click, LOGOUT]
Session for user user_2 [2025-02-01 02:31:35.0,2025-02-01 02:48:19.0) -> Events: [login, click, CHECKOUT_COMPLETE]
Session for user user_1 [2025-02-01 01:20:00.0,2025-02-01 01:35:25.0) -> Events: [login, click, click]

Conclusion

Custom triggers in Apache Flink provide a powerful mechanism to control window evaluation beyond the standard time or count policies. In this post, we saw that triggers decide when a window’s results are emitted and can fire multiple times or clear the window state. Flink’s built-in triggers (event-time, processing-time, count, etc.) cover basic needs, but they cannot be combined or tailored to complex conditions. By implementing a custom trigger, you can overcome these limitations – for example, firing early based on element count while still observing event-time boundaries, or ending a window when a domain-specific event occurs.

Key takeaways and best practices:

✅ Understand the Window Lifecycle: A trigger that fires (returns FIRE) does not close the window; the window can accumulate more data and fire again. To close it, use FIRE_AND_PURGE or design the trigger to eventually purge (as we did at the end of time windows or session gaps).

✅ Override Required Methods: At minimum override onElement, onEventTime, onProcessingTime, and clear. Use onMerge for merging window scenarios (like sessions) to handle timers or state consistently

✅ Use TriggerContext for State and Timers: If your logic needs to count events or track flags, use partitioned state via TriggerContext. Always clean up state in clear() to avoid leaks.

✅ When to Use Custom Triggers: Use them when built-in triggers don’t suffice – e.g., combined conditions (time + count), irregular event-driven firing, early and incremental results, or avoiding default behaviors (one common use case is preventing late events from retriggering a window by writing a trigger that ignores late elements). If a simple built-in trigger meets the need, stick with it for simplicity. But as shown, custom triggers shine for implementing business rules directly in the streaming logic.

You made it and reached the end. I hope you enjoyed this 👋 and happy streaming 🌊