Fix Unit test fatal fail (#10933)

Signed-off-by: xiaofan-luan <xiaofan.luan@zilliz.com>

internal/datacoord/server.go

@@ -317,7 +317,10 @@ func (s *Server) startServerLoop() {
 	s.startWatchService(s.serverLoopCtx)
 	s.startFlushLoop(s.serverLoopCtx)
 	go s.session.LivenessCheck(s.serverLoopCtx, func() {
-		log.Fatal("Data Coord disconnected from etcd, process will exit", zap.Int64("Server Id", s.session.ServerID))
+		log.Error("Data Coord disconnected from etcd, process will exit", zap.Int64("Server Id", s.session.ServerID))
+		if err := s.Stop(); err != nil {
+			log.Fatal("failed to stop server", zap.Error(err))
+		}
 	})
 }
 

internal/datanode/data_node.go

@@ -173,7 +173,10 @@ func (node *DataNode) Register() error {
 	go node.StartWatchChannels(node.ctx)
 	// Start liveness check
 	go node.session.LivenessCheck(node.ctx, func() {
-		log.Fatal("Data Node disconnected from etcd, process will exit", zap.Int64("Server Id", node.session.ServerID))
+		log.Error("Data Node disconnected from etcd, process will exit", zap.Int64("Server Id", node.session.ServerID))
+		if err := node.Stop(); err != nil {
+			log.Fatal("failed to stop server", zap.Error(err))
+		}
 	})
 
 	Params.initMsgChannelSubName()

internal/indexcoord/index_coord.go

@@ -253,7 +253,10 @@ func (i *IndexCoord) Start() error {
 		go i.watchMetaLoop()
 
 		go i.session.LivenessCheck(i.loopCtx, func() {
-			log.Fatal("Index Coord disconnected from etcd, process will exit", zap.Int64("Server Id", i.session.ServerID))
+			log.Error("Index Coord disconnected from etcd, process will exit", zap.Int64("Server Id", i.session.ServerID))
+			if err := i.Stop(); err != nil {
+				log.Fatal("failed to stop server", zap.Error(err))
+			}
 		})
 
 		startErr = i.sched.Start()

internal/indexnode/indexnode.go

@@ -185,7 +185,10 @@ func (i *IndexNode) Start() error {
 
 		//start liveness check
 		go i.session.LivenessCheck(i.loopCtx, func() {
-			log.Fatal("Index Node disconnected from etcd, process will exit", zap.Int64("Server Id", i.session.ServerID))
+			log.Error("Index Node disconnected from etcd, process will exit", zap.Int64("Server Id", i.session.ServerID))
+			if err := i.Stop(); err != nil {
+				log.Fatal("failed to stop server", zap.Error(err))
+			}
 		})
 
 		i.UpdateStateCode(internalpb.StateCode_Healthy)

internal/proxy/proxy.go

@@ -109,7 +109,10 @@ func (node *Proxy) Register() error {
 	Params.SetLogger(Params.ProxyID)
 	Params.initProxySubName()
 	go node.session.LivenessCheck(node.ctx, func() {
-		log.Fatal("Proxy disconnected from etcd, process will exit", zap.Int64("Server Id", node.session.ServerID))
+		log.Error("Proxy disconnected from etcd, process will exit", zap.Int64("Server Id", node.session.ServerID))
+		if err := node.Stop(); err != nil {
+			log.Fatal("failed to stop server", zap.Error(err))
+		}
 	})
 	// TODO Reset the logger
 	//Params.initLogCfg()

internal/querycoord/query_coord.go

@@ -183,7 +183,10 @@ func (qc *QueryCoord) Start() error {
 	go qc.watchHandoffSegmentLoop()
 
 	go qc.session.LivenessCheck(qc.loopCtx, func() {
-		log.Fatal("Query Coord disconnected from etcd, process will exit", zap.Int64("Server Id", qc.session.ServerID))
+		log.Error("Query Coord disconnected from etcd, process will exit", zap.Int64("Server Id", qc.session.ServerID))
+		if err := qc.Stop(); err != nil {
+			log.Fatal("failed to stop server", zap.Error(err))
+		}
 	})
 
 	return nil

internal/querynode/query_node.go

@@ -118,7 +118,10 @@ func (node *QueryNode) Register() error {
 	node.session.Init(typeutil.QueryNodeRole, Params.QueryNodeIP+":"+strconv.FormatInt(Params.QueryNodePort, 10), false)
 	// start liveness check
 	go node.session.LivenessCheck(node.queryNodeLoopCtx, func() {
-		log.Fatal("Query Node disconnected from etcd, process will exit", zap.Int64("Server Id", node.session.ServerID))
+		log.Error("Query Node disconnected from etcd, process will exit", zap.Int64("Server Id", node.session.ServerID))
+		if err := node.Stop(); err != nil {
+			log.Fatal("failed to stop server", zap.Error(err))
+		}
 	})
 
 	Params.QueryNodeID = node.session.ServerID

internal/rootcoord/root_coord.go

@@ -1103,6 +1103,9 @@ func (c *Core) Start() error {
 		go c.checkFlushedSegmentsLoop()
 		go c.session.LivenessCheck(c.ctx, func() {
 			log.Error("Root Coord disconnected from etcd, process will exit", zap.Int64("Server Id", c.session.ServerID))
+			if err := c.Stop(); err != nil {
+				log.Fatal("failed to stop server", zap.Error(err))
+			}
 		})
 		Params.CreatedTime = time.Now()
 		Params.UpdatedTime = time.Now()