fix(window): accommodate time shift back (#3151)

Signed-off-by: Jiyong Huang <[email protected]> (cherry picked from commit 5d16e0e) Signed-off-by: Jiyong Huang <[email protected]>
lf-edge · Sep 18, 2024 · fe84618 · fe84618
1 parent 891ce0d
commit fe84618
Showing 1 changed file with 1 addition and 1 deletion.
diff --git a/internal/topo/node/window_op.go b/internal/topo/node/window_op.go
@@ -408,7 +408,7 @@ func (o *WindowOperator) execProcessingWindow(ctx api.StreamContext, inputs []*x
 			nextTime += o.duration
 			log.Debugf("Successive tick at %v(%d), defined at %d", now, now.UnixMilli(), nextTime)
 			// If the deviation is less than 50ms, then process it. Otherwise, time may change and we'll start a new timer
-			if now.UnixMilli()-nextTime < 50 {
+			if math.Abs(float64(now.UnixMilli()-nextTime)) < 50.0 {
 				inputs = o.tick(ctx, inputs, nextTime, log)
 			} else {
 				log.Infof("Skip the tick at %v(%d) since it's too late", now, now.UnixMilli())