[Bug]: consumer subscribe error: ConsumerBusy: Exclusive consumer is already connected

[jaime0815]

Is there an existing issue for this?

• I have searched the existing issues

Environment

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

Current Behavior

2024-11-25 09:30:10.834
panic: Failed to create consumer rootcoord-dml_0, errors = server error: ConsumerBusy: Exclusive consumer is already connected

2024-11-25 09:30:10.834

2024-11-25 09:30:10.834
goroutine 11388 [running]:
2024-11-25 09:30:10.834
panic({0x51ed0c0?, 0xc02ee4bd20?})
2024-11-25 09:30:10.834
/usr/local/go/src/runtime/panic.go:1017 +0x3ac fp=0xc017155288 sp=0xc0171551d8 pc=0x1e7abec
2024-11-25 09:30:10.834
github.com/milvus-io/milvus/pkg/mq/msgstream.(*MqTtMsgStream).AsConsumer(0xc035183bd0, {0x60dc130, 0x8894f80}, {0xc019264c70?, 0x1, 0x0?}, {0xc01ba210e0, 0x4c}, 0x2)
2024-11-25 09:30:10.834
/root/milvus/pkg/mq/msgstream/mq_msgstream.go:605 +0x425 fp=0xc0171553e8 sp=0xc017155288 pc=0x3f54405
2024-11-25 09:30:10.834
github.com/milvus-io/milvus/pkg/mq/msgdispatcher.NewDispatcher({0x60dc130, 0x8894f80}, {0x79dd80e771d8, 0xc0017dd110}, 0x0, {0xc000f19780, 0x24}, 0xc00be387e0, {0xc01ba210e0, 0x4c}, ...)
2024-11-25 09:30:10.834
/root/milvus/pkg/mq/msgdispatcher/dispatcher.go:101 +0x41d fp=0xc017155878 sp=0xc0171553e8 pc=0x4718e5d
2024-11-25 09:30:10.834
github.com/milvus-io/milvus/pkg/mq/msgdispatcher.(*dispatcherManager).split.func1()
2024-11-25 09:30:10.834
/root/milvus/pkg/mq/msgdispatcher/manager.go:237 +0x17f fp=0xc017155958 sp=0xc017155878 pc=0x471e15f
2024-11-25 09:30:10.834
github.com/milvus-io/milvus/pkg/util/retry.Do({0x60dc130, 0x8894f80}, 0xc017155b88, {0xc0228edb80, 0x1, 0x0?})
2024-11-25 09:30:10.834
/root/milvus/pkg/util/retry/retry.go:44 +0x12f fp=0xc017155b10 sp=0xc017155958 pc=0x2e1024f
2024-11-25 09:30:10.834
github.com/milvus-io/milvus/pkg/mq/msgdispatcher.(*dispatcherManager).split(0xc00bf74750, 0xc00b3fc180)
2024-11-25 09:30:10.834
/root/milvus/pkg/mq/msgdispatcher/manager.go:235 +0x3da fp=0xc017155cf8 sp=0xc017155b10 pc=0x471ddba
2024-11-25 09:30:10.834
github.com/milvus-io/milvus/pkg/mq/msgdispatcher.(*dispatcherManager).Run.func1({0xc000c74e80, 0x39}, 0xc086941cc0?)
2024-11-25 09:30:10.834
/root/milvus/pkg/mq/msgdispatcher/manager.go:171 +0x2c fp=0xc017155d30 sp=0xc017155cf8 pc=0x471cf4c
2024-11-25 09:30:10.834
github.com/milvus-io/milvus/pkg/mq/msgdispatcher.(*dispatcherManager).Run.(*ConcurrentMap[...]).Range.func2({0x544c920?, 0xc00b3fc180?})
2024-11-25 09:30:10.834
/root/milvus/pkg/util/typeutil/map.go:54 +0x4b fp=0xc017155d58 sp=0xc017155d30 pc=0x471ceab
2024-11-25 09:30:10.834
sync.(*Map).Range(0xc00bf72ae0, 0xc017155e90)
2024-11-25 09:30:10.834
/usr/local/go/src/sync/map.go:476 +0x228 fp=0xc017155df0 sp=0xc017155d58 pc=0x1ed52a8
2024-11-25 09:30:10.834
github.com/milvus-io/milvus/pkg/util/typeutil.(*ConcurrentMap[...]).Range(...)

Expected Behavior

No response

Steps To Reproduce

No response

Milvus Log

No response

Anything else?

No response

[jaime0815]

Connection leak on stream when consumer creation fails.

[xiaofan-luan]

which part of the code does the leakage happened?