1

lib/streamaggr/max.go

@@ -30,7 +30,7 @@ func (as *maxAggrState) pushSamples(samples []pushSample) {
 	again:
 		v, ok := as.m.Load(outputKey)
 		if !ok {
-			fmt.Printf("wang The entry is missing in the map: %d\n", time.Now().UnixMicro())
+			// fmt.Printf("wang The entry is missing in the map: %d\n", time.Now().UnixMicro())
 			// The entry is missing in the map. Try creating it.
 			v = &maxStateValue{
 				max: s.value,
@@ -45,7 +45,7 @@ func (as *maxAggrState) pushSamples(samples []pushSample) {
 			// Use the entry created by a concurrent goroutine.
 			v = vNew
 		}
-		fmt.Printf("wang the right way %d\n", time.Now().Unix())
+		fmt.Printf("wang This is the right way %d\n", time.Now().Unix())
 		sv := v.(*maxStateValue)
 		sv.mu.Lock()
 		deleted := sv.deleted

lib/streamaggr/max_test.go

@@ -1,7 +1,7 @@
 package streamaggr
 
 import (
-	// "fmt"
+	"fmt"
 	"testing"
 	"time"
 
@@ -87,3 +87,11 @@ func TestPushSampleRace(t *testing.T) {
 	}
 	time.Sleep(1 * time.Second)
 }
+
+func TestPushSamples(t *testing.T) {
+    for i := 0; i < 5; i++ {
+        t.Run(fmt.Sprintf("Run%d", i+1), func(t *testing.T) {
+            TestPushSampleRace(t)
+        })
+    }
+}