admin/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2025-12-06 09:08:43 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: panic when streaming coord shutdown but query coord still work (#45695)

Browse Source 
issue: #44984

Signed-off-by: chyezh <chyezh@outlook.com>
This commit is contained in:
Zhen Ye 2025-11-20 11:07:06 +08:00 committed by GitHub
parent a3add6a391
commit f6411abbd7
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 5 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
internal/coordinator/snmanager/streaming_node_manager.go
View File

@ -67,6 +67,7 @@ type StreamingNodeManager struct {
cond *syncutil.ContextCond cond *syncutil.ContextCond
latestAssignments map[string]types.PChannelInfoAssigned // The latest assignments info got from streaming coord balance module. latestAssignments map[string]types.PChannelInfoAssigned // The latest assignments info got from streaming coord balance module.
nodeChangedNotifier *syncutil.VersionedNotifier // used to notify that node in streaming node manager has been changed. nodeChangedNotifier *syncutil.VersionedNotifier // used to notify that node in streaming node manager has been changed.
previousNodeIDs typeutil.UniqueSet // used to store the previous node ids.
} }
// GetBalancer returns the balancer of the streaming node manager. // GetBalancer returns the balancer of the streaming node manager.
@ -151,12 +152,15 @@ func (s *StreamingNodeManager) GetStreamingQueryNodeIDs() typeutil.UniqueSet {
} }
streamingNodes, err := balancer.GetAllStreamingNodes(context.Background()) streamingNodes, err := balancer.GetAllStreamingNodes(context.Background())
if err != nil { if err != nil {
panic(err) // when the streaming coord is on shutdown, the balancer will return an error,
// causing panic, so we need to return the previous node ids.
return s.previousNodeIDs
} }
streamingNodeIDs := typeutil.NewUniqueSet() streamingNodeIDs := typeutil.NewUniqueSet()
for _, streamingNode := range streamingNodes { for _, streamingNode := range streamingNodes {
streamingNodeIDs.Insert(streamingNode.ServerID) streamingNodeIDs.Insert(streamingNode.ServerID)
} }
s.previousNodeIDs = streamingNodeIDs
return streamingNodeIDs return streamingNodeIDs
} }