When Event Time Meets Reality: Lessons from Building Billing on Apache Flink

We thought event time would solve our billing problem.

Every usage event at Gorgias flows through Kafka and Apache Flink before becoming a billable charge. To avoid charging customers for every individual action, we aggregate usage over a 72-hour consolidation window.

The design looked straightforward: use event time, configure watermarks correctly, and let Flink do the rest.

Then historical reprocessing exposed a surprising failure mode.

Events that were perfectly ordered at the source could still arrive out of order at the billing operator after internal repartitioning. The result was overlapping consolidation windows and incorrect billing artifacts — even though watermark alignment was enabled.

This article is a detailed account of how we diagnosed the issue and what we learned about the limits of Flink’s event-time model in production.

Why Billing Requires Event-Time Semantics

Apache Flink provides a good introduction to event-time processing in its *Timely Stream Processing* documentation.

To recap the key definitions:

• Processing time refers to the system time of the machine executing the operation. In practice, this is equivalent to calling Instant.now() in Java—i.e., the time at which the event is processed by the system.
• Event time refers to the moment when the event actually occurred on the producing system. This timestamp is typically embedded within the event itself and represents the business-relevant time.

Let’s consider the following sequence of events and how they are aggregated within our three-day (72-hour) window:

As long as the system operates in real time, events are emitted and processed almost immediately. In this situation, processing time works well, as it closely matches event time.

Now, consider what happens when events are delayed due to an upstream incident or a Flink job redeployment. Some events can be delayed before reaching the pipeline. When these delayed events finally arrive, they may be aggregated together with newer, on-time events inside the same consolidation window. As a result, the computation no longer reflects the actual sequence and timing of user activity, leading to incorrect aggregations and ultimately inaccurate billing.

In this case, events 1, 2 are delayed and get aggregated with event 3, expected, but also event 4, which is wrong. And now, event 5 and 6 are aggregated together. This is ending up to 2 events billed instead of 3.

We also encountered cases where events were already being produced and stored in Kafka before the Flink job existed. When the job was finally deployed, it replayed the entire history in a matter of minutes, collapsing days of activity into a single processing-time window.

The solution to these problems is to rely on event time instead of processing time.

Event Time in a Perfect World

This theoretical model assumes a perfect world where all events flow at a steady pace through a single queue and are consumed strictly in event-time order.

This is how Flink defines the ascending-timestamps watermark strategy. By tracking the timestamp of the most recently received event, Flink can estimate the current progress in event time and decide when a consolidation window is complete.

Real-World Use Cases with Kafka

In Kafka, the “single ordered queue” assumption does not really exist. Queues are modeled as topics, and each topic is split into multiple partitions. Ordering is guaranteed only within a single partition. By choosing a partition key, we can preserve ordering for a specific dimension — for example, all events for the same customer.

A Flink job consuming a Kafka topic reads from all partitions of that topic. Each partition progresses independently, with its own sequence of events and timestamps. On top of that, the same Flink job may also consume from several topics, each with its own set of partitions.

This raises an important question: when the job receives events from many independent partitions and topics, which event time should represent the current time of the application?

Flink solves this problem by computing the global watermark as the minimum of all parallel watermarks — in other words, it always considers the oldest event-time progress across all inputs.

To maintain consistent event-time progression across all sources, I strongly recommend enabling **watermark alignment**. This is especially important because there is no guarantee that Kafka partitions or topics will be consumed at the same pace.

Watermark alignment allows Flink to coordinate event-time progress across all partitions and sources, ensuring that no stream advances too far ahead of the others beyond a configured threshold.

This becomes particularly important when replaying an entire Kafka topic. Without watermark alignment, some partitions may progress much faster than others, forcing downstream time-based operators to buffer excessive amounts of state while waiting for slower streams to catch up.

When Event-Time Ordering Breaks

There are also several scenarios where events may not arrive in event-time order. For example, after a crash, a Flink job may restart either from a checkpoint or from the committed offsets of its Kafka consumer group. Depending on the recovery point, some events may be consumed again, generating duplicates or reintroducing older event times after more recent events have already been processed. Similar situations can occur during production incidents where historical events need to be replayed manually.

In these cases, the job needs an explicit strategy for handling out-of-order events. In Flink, this is typically handled with a bounded out-of-orderness watermark strategy. The idea is to define how late an event is allowed to be. Events that arrive within this configured delay can still be processed correctly. Events that arrive after the watermark has advanced past their window are considered too late and should either be dropped or handled separately, depending on the business requirements.

The resulting architecture looks something like this:

Ingestion of Multiple Kafka Topics with Watermark Alignment and Bounded Out-of-Orderness

Without events, event time cannot advance

If the source stops producing events, time effectively stops from Flink’s perspective. Since event time advances only through incoming event timestamps, operators relying on event time — like our 72-hour consolidation window — remain stuck and are never triggered.

Other Configurations to Consider:

• Low-traffic sources One of the sources may produce only a few events per hour. Since event time advances only when new events arrive, the watermark of this source progresses very slowly and can hold back the global watermark.
• More source subtasks than Kafka partitions When the source parallelism exceeds the number of Kafka partitions, some source subtasks remain idle because they are not assigned any partition. These idle subtasks do not emit watermarks and can prevent event time from advancing.

The solution: watermark idleness To address both issues, configure watermark idleness. After a source has been inactive for a configurable period, Flink marks it as idle and temporarily excludes it from watermark computation. This allows the global watermark to continue progressing based on the active sources.

At this point, we had exhausted the toolbox provided by Flink. Watermark alignment, out-of-orderness handling, and idleness detection were all in place, allowing us to reprocess historical events while maintaining correct event-time semantics. Unfortunately, our billing use case still had one more challenge waiting for us.

Unexpected overlaps in consolidation windows

At first, the implementation appeared to be correct. Real-time events were producing accurate billing computations, and we did not observe any obvious issues in production.

The first signs of trouble appeared when we started replaying historical data. During these replays, we observed unexpected billing artifacts. After investigating several cases, we discovered that some consolidation windows were overlapping. Like this example.

In this example, Event 3 should belong to Window 1, but ended up in Window 2 because of premature closure of window 1

In the example above, Event 3 ends up in Window 2 even though, from an event-time perspective, it should still be associated with Window 1. The overlap between the two windows suggested that Window 2 had started processing before Window 1 was fully closed.

This became the central question of our investigation: how could the beginning of a consolidation window be evaluated before the previous one had completely finished?

Looking at the event timeline, we deduced that Event 3 arrived after the watermark had already advanced past the end of Window 1. As a result, Window 1 had already been evaluated and emitted by the time Event 3 was processed.

From this observation, we inferred that Flink was effectively treating Event 3 as belonging to a period that had already been considered complete. The event appeared to trigger the creation of a new consolidation window in the past, which then overlapped with the previously emitted window.

At this stage, we still did not understand why this was happening. To identify the root cause, we had to dig deeper into the implementation of the job and how event time progressed through the pipeline.

Before the consolidation window step, our job applies a deduplication step. For each customer, we store the UUID of every processed event in a MapState. This protects the pipeline against unexpected replays or duplicated events.

The consolidation step then consumes the deduplicated events and applies a 72-hour event-time window, grouped by customer and user.

Our analysis of the situation:

• During historical reprocessing, we observed overlapping consolidation windows.
• The first keyBy, based on customer, led to an uneven distribution of events across operator instances. Some slots received significantly more messages than others and therefore progressed more slowly during historical reprocessing.
• Our job applies two consecutive keyBy operations. As a result, events that eventually reach the same consolidation operator may have followed different execution paths through the job. Depending on the outcome of the first repartitioning, they may have been processed by different intermediate slots with different workloads and processing latencies before converging again at the consolidation step.
• Event time and watermark progression are evaluated at the operator/slot level rather than per key. Consequently, events sharing the same consolidation key do not progress independently in event time. A slow event can be affected by faster events processed in the same operator instance, which may advance the watermark ahead of it.
• Taken together, these observations suggest that events belonging to the same consolidation key can reach the consolidation operator with different delays and different event-time progress, despite originating from the same event-time period.

Hypothesized

• During historical reprocessing, the uneven processing introduced by the deduplication step causes some events to be delayed relative to others.
• After the second keyBy, these delayed events are redistributed and merged with streams that have already progressed further in event time.
• Because watermark alignment only operates at the source level, it is no longer sufficient to preserve event-time ordering after repartitioning.
• Some events therefore arrive at the consolidation operator in the correct order for their final keyBy partition, but after event-time progress has already moved past them. These late events can then reopen windows that Flink considers complete, leading to overlapping consolidation windows.

The deduplication step delays Event 30. After being repartitioned by the second keyBy, it is assigned to Slot 2 and merged with the events already being processed there. As a result, Event 30 arrives between Event 10 and Event 12, despite having an earlier event timestamp. From the perspective of the consolidation operator, the event stream is no longer ordered by event time.

Ideally, we would need watermark alignment not only at the source level, but also after repartitioning operations such as keyBy or any reshuffle that redistributes events across slots. This would allow Flink to preserve event-time consistency even after events are redistributed internally.

Unfortunately, this is not something Flink provides out of the box. As a result, we had to make several changes to reduce the frequency of overlapping consolidation windows rather than eliminating the problem entirely.

Avoid chaining multiple keyBys before time-based computations

In practice, this means that any processing happening before an event-time operator should use the same key as that operator. This avoids introducing an additional repartitioning step that could reorder events.

In our case, we changed the deduplication step to use the same key as the consolidation step: customer/user. This way, events stay on the same logical path before reaching the 72-hour consolidation window.

This change did not eliminate overlapping windows completely, but it reduced their occurrence by roughly 10x.

On further analysis, we realized that the root cause may be more general than our specific implementation. Any mismatch between the partitioning strategy of the source and the partitioning strategy used later in the Flink pipeline can cause events to be redistributed across different tasks. Once those events converge again downstream, differences in processing latency can introduce event-time disorder and expose the same overlapping-window behavior we observed.

At this stage, we had exhausted the toolbox provided by Flink. All the relevant watermark strategies were in place, and the remaining issues could no longer be solved through configuration alone.

Custom solution: delaying consolidation window computation during historical replays

Instead of triggering the consolidation as soon as the watermark reaches the end of the 72-hour window, we add an additional delay. This extra buffer gives slower event paths time to catch up and significantly reduces the likelihood of late events reopening windows that have already been computed.

We first investigated the extent of the problem by measuring how far overlapping windows could drift apart. In practice, we observed overlaps spanning several hours, but never more than a day.

Based on this analysis, we decided to experiment with adding a one-day buffer before evaluating the 72-hour consolidation window.

However, this approach introduced a significant drawback. The same buffer would also apply to real-time events, delaying their consolidation, billing, and availability in analytics dashboards by an additional day. For our customers, this would mean that usage data consistently lagged behind reality, which was not an acceptable trade-off.

We therefore needed a solution that could protect historical reprocessing without impacting the latency of real-time workloads.

To detect whether the job is processing real-time traffic or replaying historical data, we compare the event timestamp with the current system time. If the event timestamp is significantly older than Instant.now(), we consider that the job is no longer processing real-time events but catching up on historical data.

var eventTimestamp = value.getEventTimestamp();
var processingTime = Instant.ofEpochMilli(ctx.timerService().currentProcessingTime());
var lag = Duration.between(eventTimestamp, processingTime);
var isReprocessing = lag.compareTo(Duration.ofMinutes(15)) > 0;

For every event, we compute whether the stream is running in replay mode by comparing the event timestamp with the current system time.

When replaying historical data, the consolidation timer is registered at event timestamp + 72 hours + 1 day. For real-time traffic, the timer remains unchanged at event timestamp + 72 hours.

reprocessing historical events

The mitigation significantly reduced the problem, with roughly ten times fewer overlapping windows. Yet some windows still overlapped.

When we examined those cases in detail, we discovered that the additional buffer was not always being applied as expected.

Some timers registered by the first consolidation window could still trigger the computation of a second window before its delayed timer was reached. As a result, the second window could be evaluated earlier than intended, reintroducing the possibility of overlapping windows despite the one-day buffer.

The solution was to explicitly delete those timers once a consolidation window had been emitted. This ensured that no stale timer could later trigger the computation of another window ahead of schedule.

With this fix in place, the remaining overlapping windows disappeared.

Conclusion

What started as a seemingly straightforward event-time use case turned into a deep dive into how Apache Flink behaves under historical reprocessing workloads.

To build a reliable event-sourced billing pipeline, we first had to fully embrace Flink’s watermarking model. This meant combining several watermark strategies:

• Bounded out-of-orderness to tolerate events arriving slightly out of order.
• Watermark alignment to keep event-time progression consistent across Kafka topics and partitions consumed at different speeds.
• Idleness detection to prevent inactive sources or idle consumers from blocking event-time progress.

These configurations allowed us to correctly handle delayed events, historical replays, low-traffic topics, and multi-source ingestion scenarios.

However, we also discovered a limitation of Flink’s watermark model. Watermark alignment only applies at the source level. After a keyBy or any repartitioning operation, Flink does not preserve event-time alignment across the redistributed streams.

During large historical replays, this caused some events to arrive out of order at the consolidation operator, creating overlapping windows. We mitigated the issue by reducing repartitioning before the event-time operator, using the same key for both deduplication and consolidation, and adding custom logic to distinguish historical reprocessing from real-time traffic.

The main lesson is that configuring watermarks correctly is only part of the solution. When stateful operators, repartitioning, and historical reprocessing come into play, understanding how event time propagates through the entire execution graph becomes critical.