From 50139c994e539f2f902fc0c06ab29ee93c1bc03c Mon Sep 17 00:00:00 2001 From: sangtran Date: Sat, 12 Sep 2026 16:35:06 +0700 Subject: refactor(event-time): make source timestamps authoritative --- core/operator/watermark.go | 10 ++-------- 1 file changed, 2 insertions(+), 8 deletions(-) (limited to 'core/operator/watermark.go') diff --git a/core/operator/watermark.go b/core/operator/watermark.go index 3d591e5..43d63fa 100644 --- a/core/operator/watermark.go +++ b/core/operator/watermark.go @@ -6,20 +6,15 @@ import ( "flux/core" ) -// TimestampAssigner is responsible for event time defined by user -type TimestampAssigner[IN any] func(in IN) int64 - type WaterGeneratorOperator[IN any] struct { - assigner TimestampAssigner[IN] maxOutOfOrder int64 currentMaxTimestamp int64 lastEmittedWatermark int64 } -func NewWatermarkGeneratorOperator[IN any](maxOutOfOrder int64, assigner TimestampAssigner[IN]) *WaterGeneratorOperator[IN] { +func NewWatermarkGeneratorOperator[IN any](maxOutOfOrder int64) *WaterGeneratorOperator[IN] { return &WaterGeneratorOperator[IN]{ maxOutOfOrder: maxOutOfOrder, - assigner: assigner, lastEmittedWatermark: -1, } } @@ -34,8 +29,7 @@ func (w *WaterGeneratorOperator[IN]) Process(ctx context.Context, record core.St return err } // query the time - eventTime := w.assigner(record.Value) - record.Timestamp = eventTime + eventTime := record.Timestamp // update the max timestamp if w.currentMaxTimestamp < eventTime { -- cgit v1.3.1