Search-Click Join State: Two Buffers, One Operator, One Metric
keyed by session id, ordered by event time, first click wins, duplicates dropped by event id
SessionBuffer
searches by event time
clicks by event time
seenIds: dedup set
watermark: min(partitions)
one buffer per session id;
either side may arrive first
JoinOperator
buffer, probe other side
match: emit clicked once
expiry: emit unclicked
laggard: correct or drop
expiry is itself an output,
not the absence of one
CTR Sink
clicked / all searches
v1 value, v2 correction
meter: late_dropped_total
alert when late share grows
each search counted once;
many-to-many never inflates
probe
emit
Either order works
click before search waits in buffer;
no assumption about arrival order
Idle partition parked
watermark is the minimum, idle out;
parking is metered, never silent
CTR counts searches once
first click wins; rest are evidence;
corrections retract, never double
Why this shape
State is the join: buffer both sides by session, probe on every arrival, expire loudly. The operator never asks
which side came first, only whether the watermark still allows the answer to change.