fix: race in TumblingWindowIncAggOp FirstTimer

Protected FirstTimer access with mutex and added thread-safe getter for testing.

Signed-off-by: Jiyong Huang <huangjy@emqx.io>

Author: ngjaying

internal/topo/node/node_test/window_inc_agg_op_test.go

@@ -261,7 +261,7 @@ func TestIncAggAlignTumblingWindow(t *testing.T) {
 	time.Sleep(10 * time.Millisecond)
 	to, ok := op.WindowExec.(*node.TumblingWindowIncAggOp)
 	require.True(t, ok)
-	require.NotNil(t, to.FirstTimer)
+	require.NotNil(t, to.GetFirstTimer())
 }
 
 func TestIncAggTumblingWindow(t *testing.T) {

internal/topo/node/window_inc_agg_op.go

@@ -386,7 +386,9 @@ func (to *TumblingWindowIncAggOp) exec(ctx api.StreamContext, errCh chan<- error
 	if !EnableAlignWindow {
 		to.ticker = timex.GetTicker(to.Interval)
 	} else {
+		to.mu.Lock()
 		_, to.FirstTimer = getFirstTimer(ctx, to.windowConfig.RawInterval, to.windowConfig.TimeUnit)
+		to.mu.Unlock()
 		if to.CurrWindow == nil {
 			to.CurrWindow = newIncAggWindow(ctx, now)
 		}
@@ -465,6 +467,12 @@ outer:
 	}
 }
 
+func (to *TumblingWindowIncAggOp) GetFirstTimer() *clock.Timer {
+	to.mu.Lock()
+	defer to.mu.Unlock()
+	return to.FirstTimer
+}
+
 func (to *TumblingWindowIncAggOp) emit(ctx api.StreamContext, errCh chan<- error, now time.Time) {
 	results := &xsql.WindowTuples{
 		Content: make([]xsql.Row, 0),