fix: Move init kafka pool into once (#37786)

Introduced by #37744 Signed-off-by: Congqi Xia <[email protected]>
milvus-io · Nov 18, 2024 · 97b7ebb · 97b7ebb
1 parent 351463b
commit 97b7ebb
Showing 1 changed file with 11 additions and 9 deletions.
diff --git a/pkg/mq/msgstream/mqwrapper/kafka/pool.go b/pkg/mq/msgstream/mqwrapper/kafka/pool.go
@@ -34,15 +34,17 @@ var (
 )
 
 func initPool() {
-	pool := conc.NewPool[any](
-		hardware.GetCPUNum(),
-		conc.WithPreAlloc(false),
-		conc.WithDisablePurge(false),
-		conc.WithPreHandler(runtime.LockOSThread), // lock os thread for cgo thread disposal
-	)
-
-	kafkaCPool.Store(pool)
-	log.Info("init dynamicPool done", zap.Int("size", hardware.GetCPUNum()))
+	initOnce.Do(func() {
+		pool := conc.NewPool[any](
+			hardware.GetCPUNum(),
+			conc.WithPreAlloc(false),
+			conc.WithDisablePurge(false),
+			conc.WithPreHandler(runtime.LockOSThread), // lock os thread for cgo thread disposal
+		)
+
+		kafkaCPool.Store(pool)
+		log.Info("init dynamicPool done", zap.Int("size", hardware.GetCPUNum()))
+	})
 }
 
 // GetSQPool returns the singleton pool instance for search/query operations.