[Bug]: Datanode fails to re-watch channel if previous attempt failed due to timeout

[congqixia]

Is there an existing issue for this?

• I have searched the existing issues

Environment

- Milvus version: 293f14a
- Deployment mode(standalone or cluster): both
- MQ type(rocksmq, pulsar or kafka):    
- SDK version(e.g. pymilvus v2.0.0rc2):
- OS(Ubuntu or CentOS): 
- CPU/Memory: 
- GPU: 
- Others:

Current Behavior

When channel watch operation fails, the channel buffer is left in channel buffer manager, which will cause following attempts to watch same channel failing forever

Expected Behavior

When the data sync server failed to create the flowgraph due to timeout, it should revert the channel buffer registration

Steps To Reproduce

No response

Milvus Log

2024-05-10 10:33:47.948	
[2024/05/10 10:33:47.948 +00:00] [WARN] [datanode/data_sync_service.go:361] ["failed to register channel buffer"] [error="channel reduplicates[channel=yanliang-10kc-rootcoord-dml_14_449574292692629479v0]"]
	
2024-05-10 10:33:47.943	
[2024/05/10 10:33:47.943 +00:00] [INFO] [datanode/channel_manager.go:330] ["Start timer for ToWatch operation"] [channel=yanliang-10kc-rootcoord-dml_14_449574292692629479v0] [opID=449665233901401200] [timeout=5m0s]
	
2024-05-10 10:33:47.942	
[2024/05/10 10:33:47.942 +00:00] [INFO] [datanode/channel_manager.go:283] ["Start to execute channel operation"] [channel=yanliang-10kc-rootcoord-dml_14_449574292692629479v0] [opID=449665233901401200] [state=ToWatch]
	
2024-05-10 10:33:05.185	
[2024/05/10 10:33:05.185 +00:00] [INFO] [datanode/channel_manager.go:174] ["Fail to watch"] [opID=449665233901386957] [channel=yanliang-10kc-rootcoord-dml_14_449574292692629479v0] [State=WatchFailure]
	
2024-05-10 10:33:05.185	
[2024/05/10 10:33:05.185 +00:00] [ERROR] [msgdispatcher/client.go:82] ["register failed"] [role=datanode] [nodeID=9] [vchannel=yanliang-10kc-rootcoord-dml_14_449574292692629479v0] [error="context timeout when NewTtMsgStream"] [errorVerbose="context timeout when NewTtMsgStream\n(1) attached stack trace\n  -- stack trace:\n  | github.com/milvus-io/milvus/pkg/mq/msgstream.(*PmsFactory).NewTtMsgStream\n  | \t/go/src/github.com/milvus-io/milvus/pkg/mq/msgstream/mq_factory.go:111\n  | github.com/milvus-io/milvus/internal/util/dependency.(*DefaultFactory).NewTtMsgStream\n  | \t/go/src/github.com/milvus-io/milvus/internal/util/dependency/factory.go:144\n  | github.com/milvus-io/milvus/pkg/mq/msgdispatcher.NewDispatcher\n  | \t/go/src/github.com/milvus-io/milvus/pkg/mq/msgdispatcher/dispatcher.go:94\n  | github.com/milvus-io/milvus/pkg/mq/msgdispatcher.(*dispatcherManager).Add\n  | \t/go/src/github.com/milvus-io/milvus/pkg/mq/msgdispatcher/manager.go:93\n  | github.com/milvus-io/milvus/pkg/mq/msgdispatcher.(*client).Register\n  | \t/go/src/github.com/milvus-io/milvus/pkg/mq/msgdispatcher/client.go:76\n  | github.com/milvus-io/milvus/internal/datanode.newDmInputNode\n  | \t/go/src/github.com/milvus-io/milvus/internal/datanode/flow_graph_dmstream_input_node.go:49\n  | github.com/milvus-io/milvus/internal/datanode.getServiceWithChannel\n  | \t/go/src/github.com/milvus-io/milvus/internal/datanode/data_sync_service.go:394\n  | github.com/milvus-io/milvus/internal/datanode.newDataSyncService\n  | \t/go/src/github.com/milvus-io/milvus/internal/datanode/data_sync_service.go:503\n  | github.com/milvus-io/milvus/internal/datanode.executeWatch\n  | \t/go/src/github.com/milvus-io/milvus/internal/datanode/channel_manager.go:454\n  | github.com/milvus-io/milvus/internal/datanode.(*opRunner).watchWithTimer.func2\n  | \t/go/src/github.com/milvus-io/milvus/internal/datanode/channel_manager.go:363\n  | runtime.goexit\n  | \t/usr/local/go/src/runtime/asm_amd64.s:1598\nWraps: (2) context timeout when NewTtMsgStream\nError types: (1) *withstack.withStack (2) *errutil.leafError"] [stack="github.com/milvus-io/milvus/pkg/mq/msgdispatcher.(*client).Register\n\t/go/src/github.com/milvus-io/milvus/pkg/mq/msgdispatcher/client.go:82\ngithub.com/milvus-io/milvus/internal/datanode.newDmInputNode\n\t/go/src/github.com/milvus-io/milvus/internal/datanode/flow_graph_dmstream_input_node.go:49\ngithub.com/milvus-io/milvus/internal/datanode.getServiceWithChannel\n\t/go/src/github.com/milvus-io/milvus/internal/datanode/data_sync_service.go:394\ngithub.com/milvus-io/milvus/internal/datanode.newDataSyncService\n\t/go/src/github.com/milvus-io/milvus/internal/datanode/data_sync_service.go:503\ngithub.com/milvus-io/milvus/internal/datanode.executeWatch\n\t/go/src/github.com/milvus-io/milvus/internal/datanode/channel_manager.go:454\ngithub.com/milvus-io/milvus/internal/datanode.(*opRunner).watchWithTimer.func2\n\t/go/src/github.com/milvus-io/milvus/internal/datanode/channel_manager.go:363"]
	
	
2024-05-10 10:33:05.185	
[2024/05/10 10:33:05.185 +00:00] [INFO] [msgdispatcher/dispatcher.go:93] ["creating dispatcher..."] [pchannel=yanliang-10kc-rootcoord-dml_14] [subName=datanode-9-yanliang-10kc-rootcoord-dml_14_449574292692629479v0-false] [isMain=false]
	
2024-05-10 10:33:05.184	
[2024/05/10 10:33:05.184 +00:00] [INFO] [datanode/channel_manager.go:337] ["Stop timer for ToWatch operation timeout"] [channel=yanliang-10kc-rootcoord-dml_14_449574292692629479v0] [opID=449665233901386957] [timeout=5m0s]
	
2024-05-10 10:28:05.184	
[2024/05/10 10:28:05.184 +00:00] [INFO] [datanode/channel_manager.go:330] ["Start timer for ToWatch operation"] [channel=yanliang-10kc-rootcoord-dml_14_449574292692629479v0] [opID=449665233901386957] [timeout=5m0s]
	
2024-05-10 10:28:05.184	
[2024/05/10 10:28:05.184 +00:00] [INFO] [datanode/channel_manager.go:283] ["Start to execute channel operation"] [channel=yanliang-10kc-rootcoord-dml_14_449574292692629479v0] [opID=449665233901386957] [state=ToWatch]

Anything else?

No response