# Time and Watermarks in Confluent Cloud for Apache Flink
Timely stream processing is an extension of stateful stream processing that
incorporates time into the computation. It’s commonly used for time series
analysis, aggregations based on windows, and event processing where the time
of occurrence is important.
This topic covers event time and processing time, watermarks
and how they drive windowed operations, late data handling,
watermark alignment, and how to define time attributes in
Flink SQL.
> * [Notions of time: event time and processing time](#notions-of-time-event-time-and-processing-time)
> * [Processing time](#processing-time)
> * [Event time](#event-time)
> * [Event time and watermarks](#event-time-and-watermarks)
> * [Event-time progress](#event-time-progress)
> * [Watermarks](#watermarks)
> * [Watermark strategies](#watermark-strategies)
> * [Watermarks and timestamps](#watermarks-and-timestamps)
> * [Watermarks in parallel streams](#watermarks-in-parallel-streams)
> * [Late-arriving events](#late-arriving-events)
> * [Late data handling](#late-data-handling)
> * [Windowing](#windowing)
> * [Watermarks and windows](#watermarks-and-windows)
> * [Watermark alignment](#watermark-alignment)
> * [Time attributes](#time-attributes)
> * [Event-time attributes](#event-time-attributes)
> * [Processing-time attributes](#processing-time-attributes)
> * [Related content](#related-content)
## Notions of time: event time and processing time
Event time and processing time are the two notions of time in
stream processing. When you define windows, you choose which
notion to use.
### Processing time
Processing time refers to the system time of the machine that’s executing the
operation.
When a streaming program runs on processing time, all time-based operations,
like time windows, use the system clock of the machines that run the operator.
An hourly processing time window includes all records that arrived at a
specific operator between the times when the system clock indicated the full
hour.
For example, if an application begins running at 9:15 AM, the first hourly
processing time window includes events processed between 9:15 AM and 10:00 AM,
the next window includes events processed between 10:00 AM and 11:00 AM, and so
on.
Processing time is the simplest notion of time and requires no coordination
between streams and machines. It provides the best performance and the lowest
latency. But in distributed and asynchronous environments, processing
time doesn’t provide determinism, because it’s susceptible to:
- The speed at which records arrive in the system, for example
from a message queue.
- The speed at which records flow between operators inside the
system.
- Outages, both scheduled and unscheduled.
### Event time
Event time is the time that each individual event occurred on its producing
device. The producer typically embeds this time within the records before
they enter Flink, and Flink can extract this *event timestamp* from each
record.
In event time, the progress of time depends on the data, not on any wall
clocks. Event-time programs must specify how to generate
*event-time watermarks*, which is the mechanism that signals
progress in event time. The
[Event time and watermarks](#flink-sql-event-time-and-watermarks) section describes this
watermarking mechanism.
In a perfect world, event-time processing would yield completely consistent
and deterministic results, regardless of when events arrive, or their
ordering. But unless you know that events arrive in-order (by timestamp),
event-time processing incurs some latency while waiting for out-of-order
events. Because it’s only possible to wait for a finite period of time,
this places a limit on how deterministic event-time applications can be.
Assuming all of the data has arrived, event-time operations behave as expected,
and produce correct and consistent results even when working with out-of-order
or late events, or when reprocessing historic data.
For example, an hourly event-time window contains all records that carry an
event timestamp that falls into that hour, regardless of the order in which
they arrive, or when Flink processes them. For more information, see
[Late-arriving events](#flink-sql-event-time-lateness).
Sometimes when an event-time program is processing live data in
real-time, it uses some *processing time* operations to
guarantee that processing progresses in a timely fashion.
## Event time and watermarks
### Event-time progress
Event-time progress measures how far processing has advanced
through the event timestamp sequence, independent of wall-clock
time. A stream processor that supports *event time* needs a way
to measure this progress. For example, a window operator that
builds hourly windows needs notification when event time has
passed beyond the end of an hour, so that the operator can
close the window in progress.
Event time can progress independently of *processing time*, as measured by wall
clocks. For example, in one program, the current event time of an operator can
trail slightly behind the processing time, accounting for a delay in receiving
the events, while both proceed at the same speed. But another streaming program
might progress through weeks of event time with only a few seconds of
processing, by fast-forwarding through some historic data already buffered in
an Apache Kafka® topic.
### Watermarks
A watermark declares “all records up to this point in time have
been seen.” Watermarks are the mechanism in Flink that measures
progress in event time, determining when to advance processing
or wait for more records.
Certain SQL operations, like windows, interval joins,
time-versioned joins, and MATCH_RECOGNIZE require watermarks.
Without watermarks, they don’t produce output.
By default, every table has a watermark strategy applied. The
[SOURCE_WATERMARK](../reference/functions/datetime-functions.md#flink-sql-source-watermark-function) function provides the
default watermark strategy.
A watermark is a `long` value that usually represents epoch
milliseconds. The watermark of an operator is the minimum of
received watermarks over all partitions of all inputs. It
triggers the execution of time-based operations within this
operator before sending the watermark downstream.
Flink evaluates the watermark expression for every record, but emits only the
largest computed value downstream on a wall-clock interval, every 200 ms by
default. For more information about how Flink computes the watermark’s
timestamp value, see [Watermark strategies](#flink-sql-watermark-strategies).
The built-in function, [CURRENT_WATERMARK](../reference/functions/datetime-functions.md#flink-sql-current-watermark-function), enables
printing the current watermark for the executing operator.
Providing a timestamp is a prerequisite for providing a default watermark.
Without providing some timestamp, neither a watermark nor a time attribute
is possible.
In Flink SQL, you can use only time attributes (timestamp columns marked
with a WATERMARK clause) for time-based operations.
A time attribute must be of type `TIMESTAMP(p)` or `TIMESTAMP_LTZ(p)`, with
`0 <= p <= 3`.
Defining a watermark over a timestamp makes it a time attribute. A
**DESCRIBE** statement shows it as a **ROWTIME**.
### Watermark strategies
A watermark strategy has two related but distinct aspects: a strategy
for *when* Flink emits a watermark, and a strategy for *what timestamp
value* the watermark carries.
- **When**: Flink evaluates the watermark expression for every record,
but only emits the largest computed value downstream on a
wall-clock interval, every 200 ms by default.
- **What value**: the timestamp that Flink assigns to the watermark,
based on the `watermark_strategy_expression` in the table’s
`WATERMARK` clause. This is usually what’s meant by the shorthand
term *watermark strategy*.
Flink SQL offers these built-in strategies for calculating a
watermark’s timestamp value.
- **Default**: the [SOURCE_WATERMARK](../reference/functions/datetime-functions.md#flink-sql-source-watermark-function) function
calculates the watermark as the maximum event time seen so far in a
Kafka partition, minus a fixed out-of-orderness tolerance of 180
milliseconds.
- **Strictly ascending timestamps**: emits a watermark of the maximum
observed timestamp so far. Rows with a timestamp larger than the max
timestamp aren’t late.
- **Ascending timestamps**: emits a watermark of the maximum observed
timestamp so far, minus 1 millisecond. Rows with a timestamp larger
than or equal to the max timestamp aren’t late.
- **Bounded out-of-orderness timestamps**: emits watermarks that are
the maximum observed timestamp minus a specified delay, for example,
a 5-second delay.
If your data has out-of-orderness that exceeds the default 180 ms
tolerance, define a custom watermark strategy in a `WATERMARK`
clause. For the full syntax and an example of each strategy, see
[Watermark clause](../reference/statements/create-table.md#flink-sql-watermark-clause).
### Watermarks and timestamps
Watermarks flow as part of the data stream and carry a timestamp
*t*. A *Watermark(t)* declares that event time has reached time
*t* in that stream, meaning that there should be no more
elements from the stream with a timestamp *t’ <= t*, that is,
events with timestamps older than or equal to the watermark.
Every Kafka record has a message timestamp which is part of the
message format, and not in the payload or headers.
Timestamp semantics can be `CreateTime` (default) or
`LogAppendTime`.
The broker overwrites the timestamp only if you configure
`LogAppendTime`. Otherwise, it depends on the producer, which
means that the timestamp can be user-defined, or it can come from
the client’s clock if the user does not define it.
In most cases, a Kafka record’s timestamp is expressed in epoch
milliseconds in UTC.
The following diagram shows a stream of events with logical timestamps and
watermarks flowing inline. In this example, the events are in order with
respect to their timestamps, meaning that the watermarks are simply
periodic markers in the stream.
Watermarks are crucial for *out-of-order* streams, as shown in the
following diagram, where the events do not arrive in timestamp order. In
general, a watermark declares that by this point in the stream, all events
up to a certain timestamp should have arrived. After a watermark reaches an
operator, the operator can advance its internal *event time clock* to the
value of the watermark.
A freshly created stream element (or elements) inherits event time from
either the event that produced them or from the watermark that triggered
creation of these elements.
### Watermarks in parallel streams
Flink generates watermarks at, or directly after, source functions. Each
parallel subtask of a source function usually generates its watermarks
independently. These watermarks define the event time at that particular
parallel source.
As the watermarks flow through the streaming program, they advance the event
time at the operators where they arrive. Whenever an operator advances its
event time, it generates a new watermark downstream for its successor operators.
Some operators consume multiple input streams. For example, a union, or
operators following a *keyBy(…)* or *partition(…)* function consume multiple
input streams. Such an operator’s current event time is the minimum of its
input streams’ event times. As its input streams update their event times, so
does the operator.
The following diagram shows an example of events and watermarks flowing
through parallel streams, and operators tracking event time.
### Late-arriving events
It’s possible that certain elements violate the watermark condition, meaning
that even after the *Watermark(t)* has occurred, more elements with timestamp
*t’ <= t* occur.
In many real-world systems, arbitrary delays can affect certain elements,
making it impossible to specify a time by which all elements of a certain
event timestamp have occurred. Delaying watermarks too much causes
excessive delay in event-time window evaluation, so even when you can
bound lateness, excessive watermark delay is often not desirable.
For this reason, streaming programs can explicitly expect some *late*
elements. Late elements are elements that arrive after the system’s event
time clock, as the watermarks signal, has already passed the time of the
late element’s timestamp.
### Late data handling
Rather than relying solely on the default behavior for late elements,
Confluent Cloud for Apache Flink® provides explicit control over late-arriving events through the
`late-handling.mode` table property. Late data refers to events that arrive
after the watermark has advanced past their event timestamp.
#### Late data behavior modes
- `pass-through` (default): Late events flow through to downstream operators.
Operators like window aggregations decide whether to process or drop them.
- `filter`: Flink filters late events immediately at the source. The main
pipeline processes only on-time data, and a System Table
(`$late`) preserves the filtered events for inspection or
reprocessing.
#### System Tables
System Tables are virtual views on Flink source tables. When you configure
`late-handling.mode = 'filter'`, a System Table with the `$late` suffix
becomes available:
- Does not create new physical storage.
- Inherits schema from the source table automatically.
- Retention follows the source Kafka topic’s retention settings.
- You can query it with standard SQL (SELECT, INSERT INTO, and other
operations).
For configuration details, see [Handle Late-Arriving Data](../how-to-guides/handle-late-arriving-data.md#handle-late-arriving-data).
### Windowing
Windows scope aggregations on streams to finite time periods or
element counts. Unlike batch processing, streams are in general
infinite (unbounded), so it’s impossible to count all elements.
Instead, *windows* scope aggregates such as counts and sums,
for example *“count over the last 5 minutes”* or
*“sum of the last 100 elements”*.
Windows can be *time driven*, for example, “every 30 seconds”, or *data driven*,
for example, “every 100 elements”.
There are different types of windows, for example:
- **Tumbling windows:** no overlap.
- **Sliding windows:** with overlap.
- **Session windows:** separated by a gap of inactivity.
For more information, see:
- [Window Aggregation Queries in Confluent Cloud for Apache Flink](../reference/queries/window-aggregation.md#flink-sql-window-aggregation)
- [Window Deduplication Queries in Confluent Cloud for Apache Flink](../reference/queries/window-deduplication.md#flink-sql-window-deduplication)
- [Window Join Queries in Confluent Cloud for Apache Flink](../reference/queries/window-join.md#flink-sql-window-join)
- [Window Top-N Queries in Confluent Cloud for Apache Flink](../reference/queries/window-topn.md#flink-sql-window-top-n)
- [Windowing Table-Valued Functions (Windowing TVFs) in Confluent Cloud for Apache Flink](../reference/queries/window-tvf.md#flink-sql-window-tvfs)
### Watermarks and windows
The following example shows how watermarks propagate through a
partitioned source and trigger window computations.
In the following example, the source is a Kafka topic with 4 partitions.
The Flink job is running with a parallelism of 2, and each instance of the Kafka
source reads from 2 partitions.
Each event has a key, shown as a letter from A to D, and a timestamp.
The events shown in bold text have already been read. The events in gray, to
the left of the read position, are read next.
Flink shuffles the events that it has already read by key into the window
operators, where it counts the events by key for each hour.
Because the hour from 1 to 2 o’clock hasn’t been finalized yet, the windows
keep track of the counters for that hour. There have been two events for key A
for that hour, one event for key B, and so on.
Because events for the following hour have already begun to appear, these
windows also maintain counters for the hour from 2 o’clock to 3 o’clock.
These windows wait for watermarks to trigger them to produce their results. The
watermarks come from the watermark generators in the Kafka source operators.
For each Kafka partition, the watermark generator keeps track of the largest
timestamp seen so far, and subtracts from that an estimate of the expected
out-of-orderness.
For example, for Partition 1, the largest timestamp is 1:30. Assuming that the
events are at most 1 minute out of order, then the watermark for Partition 1 is
1:29.
A similar computation for Partition 3 yields a watermark of 1:30, and so on for
the remaining partitions.
Each of the two Kafka source instances takes as its watermark the minimum of
these per-partition watermarks.
From the point of view of the uppermost Kafka source operator, the watermark it
produces should include a timestamp that reflects how complete the stream is
that it is producing.
This stream from Kafka Source 1 includes events from both Partition 1 and
Partition 3, so it can be no more complete than the furthest behind of these
two partitions, which is Partition 1.
Although Partition 1 has seen an event with a timestamp as late as 1:30, it
reports its watermark as 1:29, because it allows for its events to be up to
one minute out-of-order.
The same reasoning applies as the watermarks flow downstream through the
job graph. Each instance of the window operator has received watermarks
from the two Kafka source instances.
The current watermark at both of the window operators is 1:17, because this is
the furthest behind of the watermarks coming into the windows from the Kafka
sources.
The furthest behind of all four Kafka partitions determines the overall progress
of the windows.
### Watermark alignment
Watermark alignment enables you to specify how tightly synchronized your
streams should be, preventing any of the sources from getting too far ahead of
the others. It addresses the problem of temporal joins between streams with
progressively diverging timestamps.
When performing temporal joins between two streams, if one stream is
significantly ahead of the other, Flink must buffer data from the leading
stream while waiting for the watermark of the lagging stream to advance.
As timestamps diverge further, the buffering requirements grow, potentially
causing performance degradation and operational issues, such as
checkpointing failures.
Watermark alignment enables you to pause reading from streams that are too far
ahead, enabling lagging streams to catch up and preventing the situation from
worsening. This feature is particularly valuable when joining streams that have
naturally diverging timestamps, such as when one data source produces events
more frequently or with different timing characteristics than another.
Watermark alignment provides these benefits:
- Reduces memory buffering requirements.
- Improves performance by preventing excessive data buffering.
- Prevents operational problems such as checkpointing failures.
- Provides control over stream synchronization.
Confluent Cloud for Apache Flink enables watermark alignment by default.
Set the `sql.tables.scan.watermark-alignment.max-allowed-drift` session
option to change the maximum allowed deviation, or *watermark drift*.
The default maximum watermark drift is 5 minutes. This value matches the
default maximum idleness detection timeout, which is also 5 minutes.
Otherwise, watermark alignment would occur while Flink waits for a
partition to switch to idle, potentially wasting CPU resources.
Only increase the watermark alignment’s maximum allowed drift to match the
idleness timeout when you increase the idleness timeout. For more
information, see [Validate or disable idleness handling](../operate-and-deploy/best-practices.md#flink-sql-best-practices-for-statements-idleness).
Decrease the maximum allowed drift when both of these conditions
are true:
- Record throughput, expressed as records per minute of event
time, is too high for windowed or temporal operators to buffer
five minutes of data.
- The window length is less than five minutes.
## Time attributes
Confluent Cloud for Apache Flink can process data based on different notions of time.
- **Event time** refers to stream processing based on timestamps that are
attached to each row. The timestamps can encode when an event happened.
- **Processing time** refers to the machine’s system time that’s
executing the operation. Processing time is also known as “epoch time”,
for example, Java’s `System.currentTimeMillis()`. Processing time is
not supported in Confluent Cloud for Apache Flink.
Time attributes can be part of every table schema. You define them when you
create a table from a `CREATE TABLE` DDL statement.
After you define a time attribute, you can reference it as a field and use
it in time-based operations. As long as a query does not modify a time
attribute, and simply forwards it from one part of a query to another, it
remains a valid time attribute.
Time attributes behave like regular timestamps and are accessible for
calculations. When used in calculations, Flink materializes time attributes
and they act as standard timestamps, but you can’t use ordinary timestamps
in place of, or convert them to, time attributes.
### Event-time attributes
An event-time attribute is a table column marked with a
WATERMARK clause that enables time-based operations on event
timestamps. Event time enables a table program to produce
results based on timestamps in every record, which allows for
consistent results despite out-of-order or late events. Event
time also ensures the replayability of the results of the table
program when reading records from persistent storage.
Also, event time enables unified syntax for table programs in both batch and
streaming environments. A time attribute in a streaming environment can be a
regular column of a row in a batch environment.
To handle out-of-order events and to distinguish between on-time and late
events in streaming, Flink must know the timestamp for each row, and it also
needs regular indications of how far along in event time the processing has
progressed so far, by using
[watermarks](#flink-sql-event-time-and-watermarks).
You can define event-time attributes in
[CREATE TABLE](../reference/statements/create-table.md#flink-sql-create-table) statements.
#### Define event-time attributes in DDL
You define the event-time attribute by using a WATERMARK clause in a
`CREATE TABLE` DDL statement. A watermark statement defines a watermark
generation expression on an existing event-time field, which marks the
event-time field as the event-time attribute. For more information about
watermark strategies, see [Watermark clause](../reference/statements/create-table.md#flink-sql-watermark-clause).
Flink SQL supports defining an event-time attribute on TIMESTAMP and
TIMESTAMP_LTZ columns. If the source represents the timestamp data as
year-month-day-hour-minute-second, usually a string value without
time-zone information, for example, `2020-04-15 20:13:40.564`, define
the event-time attribute as a `TIMESTAMP` column.
```sql
CREATE TABLE user_actions (
user_name STRING,
data STRING,
user_action_time TIMESTAMP(3),
-- Declare the user_action_time column as an event-time attribute
-- and use a 5-seconds-delayed watermark strategy.
WATERMARK FOR user_action_time AS user_action_time - INTERVAL '5' SECOND
) WITH (
...
);
SELECT TUMBLE_START(user_action_time, INTERVAL '10' MINUTE), COUNT(DISTINCT user_name)
FROM user_actions
GROUP BY TUMBLE(user_action_time, INTERVAL '10' MINUTE);
```
If the source represents the timestamp data as epoch time, which is
usually a LONG value such as `1618989564564`, consider defining the
event-time attribute as a `TIMESTAMP_LTZ` column.
```sql
CREATE TABLE user_actions (
user_name STRING,
data STRING,
ts BIGINT,
time_ltz AS TO_TIMESTAMP_LTZ(ts, 3),
-- Declare the time_ltz column as an event-time attribute
-- and use a 5-seconds-delayed watermark strategy.
WATERMARK FOR time_ltz AS time_ltz - INTERVAL '5' SECOND
) WITH (
...
);
SELECT TUMBLE_START(time_ltz, INTERVAL '10' MINUTE), COUNT(DISTINCT user_name)
FROM user_actions
GROUP BY TUMBLE(time_ltz, INTERVAL '10' MINUTE);
```
### Processing-time attributes
Processing time enables a table program to produce results based on the
time of the local machine. It’s the simplest notion of time, but it
generates non-deterministic results. Processing time doesn’t require
timestamp extraction or watermark generation.
Processing time is not supported in Confluent Cloud for Apache Flink.
## Related content
Flink implements many techniques from the Dataflow Model. For a good
introduction to event time and watermarks, see these articles and videos.
- Course: [Watermarks Demystified](https://developer.confluent.io/courses/flink-sql/watermarks/)
- Video: [Watermark Alignment Explained in 2 Minutes](https://www.youtube.com/watch?v=HxVcs9SNZEg)
- Blog post: [Introducing Stream Windows in Apache Flink](https://flink.apache.org/news/2015/12/04/Introducing-windows.html)
- [Streaming 101 (O’Reilly online learning)](https://www.oreilly.com/ideas/the-world-beyond-batch-streaming-101) by Tyler Akidau
- [Dataflow Model](https://research.google.com/pubs/archive/43864.pdf)
- [CREATE TABLE Statement in Confluent Cloud for Apache Flink](../reference/statements/create-table.md#flink-sql-create-table)
- [Flink SQL Queries](../reference/queries/overview.md#flink-sql-queries)
- [Determinism](determinism.md#flink-sql-determinism)
- [Validate your watermark strategy](../operate-and-deploy/best-practices.md#flink-sql-best-practices-for-statements-validate-watermarks)
- [Validate or disable idleness handling](../operate-and-deploy/best-practices.md#flink-sql-best-practices-for-statements-idleness)
- [Handle Late-Arriving Data](../how-to-guides/handle-late-arriving-data.md#handle-late-arriving-data)
#### NOTE
This website includes content developed at the [Apache Software Foundation](https://www.apache.org/)
under the terms of the [Apache License v2](https://www.apache.org/licenses/LICENSE-2.0.html).