Apache Flink is a native stream-processing engine built around continuous dataflow, stateful operators, checkpointed recovery, and event-time semantics. Unlike engines that primarily run repeated mini-batches, Flink operators process records as they flow through the topology. Bounded batch jobs are treated as a special case of the same execution model, while unbounded streams are the default case the system is designed to handle.
The most important idea in Flink is that correctness should be defined in event time, not merely in the time at which a worker happens to see a record. A purchase that occurred at 12:03 but reached the pipeline at 12:09 usually belongs in the 12:00-12:05 window. Processing-time windows would count it in the wrong interval. Event-time processing lets the application compute answers according to when facts happened in the business domain, even when networks, devices, brokers, and retries reorder delivery.
Watermarks are Flink’s mechanism for advancing event time. A watermark is a progress signal that says the source believes it has seen all, or almost all, events up to a timestamp. Operators use watermarks to close windows, fire timers, clean up state, and identify late records. The signal is intentionally explicit because unbounded data never naturally ends. A streaming system must decide when a partial result is useful enough to emit and how long to hold state for corrections.
This topic fills the gap between broad stream-processing vocabulary and production Flink design. It focuses on event time versus processing time, watermark generation, late-event handling, keyed state, checkpoint barriers, exactly-once state consistency, and the operational traps that appear when a clean tutorial meets uneven partitions and real backpressure.
Event time is the timestamp embedded in the event itself: the time an order was placed, a device measurement was taken, or a user action occurred. It is the right time domain for billing, analytics, anomaly detection, sessionization, and legal audit trails because those outputs depend on the domain fact, not on ingestion luck. In Flink, event-time programs assign timestamps to records and use those timestamps for windows and timers.
Processing time is the wall-clock time on the machine executing the operator. It is simple and low-overhead, but it mixes domain behavior with infrastructure behavior. Network delays, broker lag, rebalances, and overloaded workers change processing time without changing the underlying user activity. Processing-time windows are still useful for operational metrics such as “records processed per minute by this job,” but they are usually wrong for business facts that arrive late.
A watermark with timestamp T means the system’s event-time clock can advance to T. For an operator, any record with an event timestamp less than the current watermark is late. Watermarks can be exact when a bounded input is fully known, but production stream sources usually use heuristics such as bounded out-of-orderness, per-partition source progress, or idleness detection.
Windows turn an infinite stream into finite regions. Tumbling windows group non-overlapping intervals, sliding windows overlap, and session windows merge events separated by short gaps. Event-time timers provide lower-level control: a keyed operator can register a callback for a future event timestamp and Flink will fire it when the watermark passes that timestamp.
A watermark is not a magic proof that no older event exists. It is a progress contract between the source and downstream operators. The tighter the watermark, the lower the latency and the higher the late-event risk; the more conservative the watermark, the more state and output delay the job carries.
A Flink source or user function extracts event timestamps and emits watermarks. The common bounded-out-of-orderness strategy delays watermarks behind the maximum event timestamp seen so far. If the maximum observed event time is 12:10:00 and the out-of-orderness bound is 30 seconds, the watermark advances to 12:09:30. Events earlier than that point are treated as late.
WatermarkStrategy<OrderEvent> strategy =
WatermarkStrategy
.<OrderEvent>forBoundedOutOfOrderness(Duration.ofSeconds(30))
.withTimestampAssigner((event, previousTimestamp) -> event.eventTimeMillis())
.withIdleness(Duration.ofMinutes(2));
DataStream<OrderEvent> orders =
env.fromSource(kafkaSource, strategy, "orders");
DataStream<RevenueByWindow> hourlyRevenue =
orders
.keyBy(OrderEvent::merchantId)
.window(TumblingEventTimeWindows.of(Time.hours(1)))
.allowedLateness(Time.minutes(10))
.sideOutputLateData(LATE_ORDERS)
.aggregate(new RevenueAggregate(), new RevenueWindowFunction());Watermarks flow through the same operator graph as records. For operators with multiple inputs or multiple upstream partitions, event time can only advance as fast as the slowest active input. This preserves correctness because a join or union cannot safely close a window while one input might still deliver older records. It also means one idle Kafka partition can stall every downstream event-time window unless the source marks it idle.
Keyed process functions can register event-time timers for custom logic that does not fit a built-in window. A fraud detector might keep per-account state, emit an alert if enough suspicious events arrive within ten minutes, and clear state when the watermark passes the session deadline.
class BurstDetector
extends KeyedProcessFunction<String, PaymentEvent, Alert> {
private transient ValueState<BurstState> state;
public void processElement(
PaymentEvent event,
Context ctx,
Collector<Alert> out) throws Exception {
BurstState current = state.value();
if (current == null) {
long deadline = event.eventTimeMillis() + Duration.ofMinutes(10).toMillis();
current = new BurstState(0, event.amountCents(), deadline);
ctx.timerService().registerEventTimeTimer(deadline);
}
current.count += 1;
current.totalCents += event.amountCents();
state.update(current);
if (current.count >= 5 && current.totalCents > 100_000) {
out.collect(Alert.highVelocity(event.accountId(), current.totalCents));
}
}
public void onTimer(long timestamp, OnTimerContext ctx, Collector<Alert> out)
throws Exception {
state.clear();
}
}Flink snapshots state with distributed checkpoint barriers. Sources inject barriers into streams. Operators align barriers from their inputs, snapshot their keyed and operator state, forward the barrier, and continue. After every operator has persisted its snapshot, the checkpoint is complete and recovery can restart from the checkpointed source offsets and state. Flink’s blog on aligned and unaligned checkpoints notes that barriers can align in milliseconds in healthy pipelines, but backpressure makes barriers travel slowly and increases checkpoint duration.
Billing systems need to count usage in the period when it occurred. Event-time windows prevent delayed mobile, edge, or partner data from being charged in the wrong interval. Allowed lateness gives finance teams a controlled correction window, and side outputs for very late data provide an audit path instead of silent loss.
Fraud patterns often depend on time relationships: many cards used from one account in ten minutes, repeated failed login events followed by a password reset, or geographic impossibilities within a travel window. Flink’s keyed state and event-time timers let the detector reason about those patterns without assuming records arrive in order.
Session windows are a natural fit for websites and applications where a user journey is defined by inactivity gaps. Late events can merge sessions that looked separate when early results were emitted. That means sinks must support updates or compensation records, and dashboards must distinguish provisional panes from finalized panes.
Flink is commonly used to join fast event streams with reference data, compute rolling aggregates, and feed downstream systems such as search indexes, feature stores, and alerting pipelines. The state model is useful when the enrichment table is too large for every operator to hold in memory but still needs local, keyed access with checkpointed recovery.
Do not pick a watermark delay by taste. Measure the distribution of ingestion delay by source, partition, region, and event type. A 30-second lag may work for broker-local events and fail for mobile clients that reconnect after flights. A 30-minute lag may preserve more late data but make dashboards and state stores too slow. Production pipelines should export percentiles for event-time delay and late-record counts.
lateness_ms = processing_time_ms - event_time_ms
watermark_lag_seconds = {
"p50": percentile(lateness_ms, 50) / 1000,
"p95": percentile(lateness_ms, 95) / 1000,
"p99": percentile(lateness_ms, 99) / 1000,
"max_observed": max(lateness_ms) / 1000
}
if late_records_per_minute > error_budget:
increase_out_of_orderness_bound()
elif output_latency_p95 > product_budget:
decrease_out_of_orderness_bound_or_emit_early_results()Side outputs for late data are not just a debugging convenience. They are how the pipeline remains auditable when a record arrives after state has been cleaned. Late-data topics can feed reconciliation jobs, human review, or compensation workflows. Without them, an aggressive watermark becomes an invisible data-loss policy.
Exactly-once state inside Flink does not automatically mean exactly-once side effects in every external system. Sinks need two-phase commit support, idempotent upserts keyed by window and entity, or replay-aware deduplication. For a window aggregate, a stable key such as `(metric_name, window_start, window_end, entity_id)` should identify the logical result across retries.
Historical backfills often have perfect or near-perfect input knowledge and can tolerate high throughput over low latency. Live streams need bounded delay and operational observability. Keeping the same business transform but changing source, watermark, parallelism, and sink settings is usually safer than forcing one runtime profile to serve both workloads.