1 files changed, 2 insertions, 8 deletions

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 {