fix(test): resolve data races in redis mock and nexmark

1. Fix data race in pkg/mock/test_source.go used by Redis tests.

2. Fix data race in global slice access in internal/io/nexmark/event.go.

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

Author: ngjaying

internal/io/nexmark/event.go

@@ -18,6 +18,7 @@ import (
 	"fmt"
 	"math/rand"
 	"strings"
+	"sync"
 )
 
 var (
@@ -29,6 +30,7 @@ var (
 	PersonIDs       []uint64
 	AuctionIDs      []uint64
 	categoriesCount int
+	mu              sync.RWMutex
 )
 
 func init() {
@@ -68,11 +70,15 @@ func (p Person) ToMap() map[string]interface{} {
 
 func genPersonID(r *rand.Rand) uint64 {
 	id := uint64(r.Int())
+	mu.Lock()
+	defer mu.Unlock()
 	PersonIDs = append(PersonIDs, id)
 	return id
 }
 
 func pickPersonID(r *rand.Rand) uint64 {
+	mu.RLock()
+	defer mu.RUnlock()
 	return PersonIDs[r.Int()%len(PersonIDs)]
 }
 
@@ -152,11 +158,15 @@ func NewAuction(eventID int64, time uint64) Auction {
 
 func genAuctionID(r *rand.Rand) uint64 {
 	id := uint64(r.Int())
+	mu.Lock()
+	defer mu.Unlock()
 	AuctionIDs = append(AuctionIDs, id)
 	return id
 }
 
 func pickAuctionID(r *rand.Rand) uint64 {
+	mu.RLock()
+	defer mu.RUnlock()
 	return AuctionIDs[r.Int()%len(AuctionIDs)]
 }
 

pkg/mock/test_source.go

@@ -149,9 +149,11 @@ func TestSourceConnectorCompare(t *testing.T, r api.Source, props map[string]any
 				}
 			}()
 		case api.BytesSource:
-			err = ss.Subscribe(ctx, ingestBytes, ingestErr)
+			err := ss.Subscribe(ctx, ingestBytes, ingestErr)
+			assert.NoError(t, err)
 		case api.TupleSource:
-			err = ss.Subscribe(ctx, ingestTuples, ingestErr)
+			err := ss.Subscribe(ctx, ingestTuples, ingestErr)
+			assert.NoError(t, err)
 		default:
 			panic("wrong source type")
 		}