Documentation
¶
Overview ¶
Package watermark provides watermark generation for event-time processing.
A watermark is a timestamp that says "no records with event time < T will arrive after this point." Watermarks flow through the stream as special Records (IsWatermark == true) and are used by the WindowOperator to decide when a window is complete.
Watermarks are generated by the Source. The simplest strategy is to use the maximum timestamp seen so far minus an allowed lateness bound. This is the BoundedOutOfOrderness generator — the most common strategy.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type BoundedOutOfOrderness ¶
type BoundedOutOfOrderness struct {
// contains filtered or unexported fields
}
BoundedOutOfOrderness is the most common watermark generator strategy. It tracks the maximum timestamp seen and subtracts a fixed lateness bound. The resulting watermark is: max_timestamp - allowed_lateness.
This handles out-of-order events: if allowed_lateness is 5 seconds, any record arriving up to 5 seconds late (relative to the max timestamp) will still be placed in its correct window.
Example:
Records arrive: ts=10, ts=8, ts=15, ts=12 Max timestamp seen: 15 Allowed lateness: 5s Current watermark: 15 - 5 = 10 → Windows with end time <= 10 can now close
func NewBoundedOutOfOrderness ¶
func NewBoundedOutOfOrderness(allowedLateness time.Duration) *BoundedOutOfOrderness
NewBoundedOutOfOrderness creates a watermark generator with the given allowed lateness. Events up to this duration late will still be placed in their correct window.
func (*BoundedOutOfOrderness) CurrentWatermark ¶
func (g *BoundedOutOfOrderness) CurrentWatermark() time.Time
CurrentWatermark returns the current watermark timestamp. This is the same as GetWatermark — it's here to satisfy the interface.
func (*BoundedOutOfOrderness) GetWatermark ¶
func (g *BoundedOutOfOrderness) GetWatermark() time.Time
GetWatermark returns the current watermark: max_timestamp - allowed_lateness. If no records have been seen, returns the zero time.
func (*BoundedOutOfOrderness) OnRecord ¶
func (g *BoundedOutOfOrderness) OnRecord(timestamp time.Time)
OnRecord updates the maximum timestamp seen.
type WatermarkGenerator ¶
type WatermarkGenerator interface {
// OnRecord updates the generator's internal state with the record's timestamp.
OnRecord(timestamp time.Time)
// GetWatermark returns the current watermark timestamp.
// If no records have been seen, returns time.Time{} (zero value).
GetWatermark() time.Time
// CurrentWatermark returns the current watermark as a special Record
// with IsWatermark=true. Returns the zero Record if no watermark is ready.
CurrentWatermark() time.Time
}
WatermarkGenerator creates watermark Records based on the timestamps of data records flowing through the stream.
The Source calls OnRecord for each data record, and periodically calls GetWatermark to check if a new watermark should be emitted.
Source Files