milvus-io
/
milvus
mirror of https://github.com/milvus-io/milvus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

enhance: Graceful stop flowgraph manager when stopping datanode (#36229) 

Flowgraph manager is not stopped durong datanode stopping procedure
which may lead to unexpect flowgraph behavior during/after datanode stop
progress.

---------

Signed-off-by: Congqi Xia <congqi.xia@zilliz.com>
pull/36291/head
congqixia 2024-09-14 15:55:08 +08:00 committed by GitHub
parent b8b4aea4f5
commit 3352030a84
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 17 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
internal/datanode/data_node.go
View File

@ -362,6 +362,11 @@ func (node *DataNode) Stop() error {
node.channelManager.Close()
}
if node.flowgraphManager != nil {
node.flowgraphManager.ClearFlowgraphs()
node.flowgraphManager.Close()
}
if node.writeBufferManager != nil {
node.writeBufferManager.Stop()
}

12
internal/datanode/services_test.go
View File

@ -685,6 +685,8 @@ func (s *DataNodeServicesSuite) TestSyncSegments() {
mockFlowgraphManager := pipeline.NewMockFlowgraphManager(s.T())
mockFlowgraphManager.EXPECT().GetFlowgraphService(mock.Anything).
Return(pipeline.NewDataSyncServiceWithMetaCache(cache), true)
mockFlowgraphManager.EXPECT().ClearFlowgraphs().Return().Maybe()
mockFlowgraphManager.EXPECT().Close().Return().Maybe()
s.node.flowgraphManager = mockFlowgraphManager
ctx := context.Background()
req := &datapb.SyncSegmentsRequest{
@ -783,6 +785,8 @@ func (s *DataNodeServicesSuite) TestSyncSegments() {
mockFlowgraphManager := pipeline.NewMockFlowgraphManager(s.T())
mockFlowgraphManager.EXPECT().GetFlowgraphService(mock.Anything).
Return(pipeline.NewDataSyncServiceWithMetaCache(cache), true)
mockFlowgraphManager.EXPECT().ClearFlowgraphs().Return().Maybe()
mockFlowgraphManager.EXPECT().Close().Return().Maybe()
s.node.flowgraphManager = mockFlowgraphManager
ctx := context.Background()
req := &datapb.SyncSegmentsRequest{
@ -869,6 +873,8 @@ func (s *DataNodeServicesSuite) TestSyncSegments() {
mockFlowgraphManager := pipeline.NewMockFlowgraphManager(s.T())
mockFlowgraphManager.EXPECT().GetFlowgraphService(mock.Anything).
Return(pipeline.NewDataSyncServiceWithMetaCache(cache), true)
mockFlowgraphManager.EXPECT().ClearFlowgraphs().Return().Maybe()
mockFlowgraphManager.EXPECT().Close().Return().Maybe()
s.node.flowgraphManager = mockFlowgraphManager
ctx := context.Background()
req := &datapb.SyncSegmentsRequest{
@ -966,6 +972,8 @@ func (s *DataNodeServicesSuite) TestSyncSegments() {
mockFlowgraphManager := pipeline.NewMockFlowgraphManager(s.T())
mockFlowgraphManager.EXPECT().GetFlowgraphService(mock.Anything).
Return(pipeline.NewDataSyncServiceWithMetaCache(cache), true)
mockFlowgraphManager.EXPECT().ClearFlowgraphs().Return().Maybe()
mockFlowgraphManager.EXPECT().Close().Return().Maybe()
s.node.flowgraphManager = mockFlowgraphManager
ctx := context.Background()
req := &datapb.SyncSegmentsRequest{
@ -1046,6 +1054,8 @@ func (s *DataNodeServicesSuite) TestSyncSegments() {
mockFlowgraphManager := pipeline.NewMockFlowgraphManager(s.T())
mockFlowgraphManager.EXPECT().GetFlowgraphService(mock.Anything).
Return(pipeline.NewDataSyncServiceWithMetaCache(cache), true)
mockFlowgraphManager.EXPECT().ClearFlowgraphs().Return().Maybe()
mockFlowgraphManager.EXPECT().Close().Return().Maybe()
s.node.flowgraphManager = mockFlowgraphManager
ctx := context.Background()
req := &datapb.SyncSegmentsRequest{
@ -1104,6 +1114,8 @@ func (s *DataNodeServicesSuite) TestSyncSegments() {
mockFlowgraphManager := pipeline.NewMockFlowgraphManager(s.T())
mockFlowgraphManager.EXPECT().GetFlowgraphService(mock.Anything).
Return(pipeline.NewDataSyncServiceWithMetaCache(cache), true)
mockFlowgraphManager.EXPECT().ClearFlowgraphs().Return().Maybe()
mockFlowgraphManager.EXPECT().Close().Return().Maybe()
s.node.flowgraphManager = mockFlowgraphManager
ctx := context.Background()
req := &datapb.SyncSegmentsRequest{