Stream processing groups events by their OWN timestamp (event time), not by when they happen to arrive at the engine -- a mobile client might upload a batch of readings minutes after they were recorded, and they should still land in the window they actually belong to. A WATERMARK is the engine's answer to "how do I know a window is done waiting for more events, if events can arrive late and out of order?" -- it's a monotonically advancing estimate of "we won't wait for anything older than this anymore," computed as the newest timestamp seen so far minus an allowed out-of-orderness slack.

Implement:

• TumblingWindowAssigner.assign_window(timestamp) -- the start of the fixed-size window timestamp falls into: timestamp - (timestamp % window_size). Windows are [start, start + window_size).
• WatermarkGenerator.on_event(timestamp) -- observe a timestamp, advancing max_timestamp_seen ONLY if this one is newer than anything seen before (an out-of-order event must never move it backward).
• WatermarkGenerator.current_watermark() -- max_timestamp_seen - max_out_of_orderness, or float("-inf") if nothing has been observed yet (so the very first event is never treated as late).

WindowManager.add_event is given -- it checks a new event's lateness against the watermark computed from events seen BEFORE it (not itself), observes the new event either way, then either buckets it into its window ("ON_TIME") or routes it to late_events ("LATE") if its window has already closed (window_end <= watermark).

Example

assigner = TumblingWindowAssigner(window_size=10)
assigner.assign_window(2)    # -> 0
assigner.assign_window(25)   # -> 20

wm = WatermarkGenerator(max_out_of_orderness=5)
wm.on_event(20)
wm.current_watermark()       # -> 15
wm.on_event(12)               # out of order -- older than 20
wm.current_watermark()       # -> still 15, unchanged

Constraints

• on_event must be a one-directional ratchet: max_timestamp_seen only ever grows.
• Before any event, current_watermark() is float("-inf") -- there is no such thing as a "late" first event.
• Lateness is exclusive at the boundary: a window is closed (its events are late) once window_end <= watermark, not <.

Real-world grounding

• Flink's tumbling event-time window assigner: flink-runtime/src/main/java/org/apache/flink/streaming/api/windowing/assigners/TumblingEventTimeWindows.java
• Flink's bounded-out-of-orderness watermark generator (the exact maxTimestampSeen - maxOutOfOrderness strategy this exercise implements): flink-core/src/main/java/org/apache/flink/api/common/eventtime/BoundedOutOfOrdernessWatermarks.java

Real Flink emits watermarks periodically (not on every single event) and supports routing late events to an explicit side output stream for separate handling rather than just a dropped/collected list -- this exercise implements the underlying lateness decision both are built on.