milvus: [Bug]: [nightly] Panic occurs reporting "Failed to create consumer by-dev-rootcoord-dml_48" during flush

Is there an existing issue for this?

• I have searched the existing issues

Environment

- Milvus version: 440bce9
- Deployment mode(standalone or cluster): cluster
- SDK version(e.g. pymilvus v2.0.0rc2): pymilvus dev 69
- OS(Ubuntu or CentOS): 
- CPU/Memory: 
- GPU: 
- Others:

Current Behavior

Panic occurs reporting “Failed to create consumer by-dev-rootcoord-dml_48” during flush:

mdp-587-n-milvus-datacoord-5bd8cdb979-r2znf_milvus-ci_datacoord-06c98568778f02f95f95c464236ecda64aadd2c71419c32cef243bccee80e769.log:[2022/06/11 19:11:40.043 +00:00] [FATAL] [logutil.go:134] [panic] [recover="\"Failed to create consumer by-dev-rootcoord-dml_48, error = All attempts results:\\nattempt #1:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #2:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #3:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #4:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #5:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #6:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #7:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #8:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #9:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #10:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #11:server error: ConsumerBusy: Exclusive consumer is already connected\\n\""] [stack="github.com/milvus-io/milvus/internal/util/logutil.LogPanic\n\t/go/src/github.com/milvus-io/milvus/internal/util/logutil/logutil.go:134\nruntime.gopanic\n\t/usr/local/go/src/runtime/panic.go:965\ngithub.com/milvus-io/milvus/internal/mq/msgstream.(*mqMsgStream).AsConsumerWithPosition\n\t/go/src/github.com/milvus-io/milvus/internal/mq/msgstream/mq_msgstream.go:174\ngithub.com/milvus-io/milvus/internal/mq/msgstream.(*mqMsgStream).AsConsumer\n\t/go/src/github.com/milvus-io/milvus/internal/mq/msgstream/mq_msgstream.go:131\ngithub.com/milvus-io/milvus/internal/mq/msgstream.UnsubscribeChannels\n\t/go/src/github.com/milvus-io/milvus/internal/mq/msgstream/msgstream_util.go:35\ngithub.com/milvus-io/milvus/internal/datacoord.(*ChannelManager).CleanupAndReassign\n\t/go/src/github.com/milvus-io/milvus/internal/datacoord/channel_manager.go:781\ngithub.com/milvus-io/milvus/internal/datacoord.(*ChannelManager).processAck\n\t/go/src/github.com/milvus-io/milvus/internal/datacoord/channel_manager.go:621\ngithub.com/milvus-io/milvus/internal/datacoord.(*ChannelManager).watchChannelStatesLoop\n\t/go/src/github.com/milvus-io/milvus/internal/datacoord/channel_manager.go:655"]
mdp-587-n-milvus-datacoord-5bd8cdb979-r2znf_milvus-ci_datacoord-1ec4b7103808e14df08c854e5f9569a561a6d4b8b0aa3fb8eab51a07ba974082.log:[2022/06/11 19:15:33.591 +00:00] [FATAL] [logutil.go:134] [panic] [recover="\"Failed to create consumer by-dev-rootcoord-dml_48, error = All attempts results:\\nattempt #1:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #2:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #3:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #4:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #5:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #6:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #7:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #8:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #9:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #10:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #11:server error: ConsumerBusy: Exclusive consumer is already connected\\n\""] [stack="github.com/milvus-io/milvus/internal/util/logutil.LogPanic\n\t/go/src/github.com/milvus-io/milvus/internal/util/logutil/logutil.go:134\nruntime.gopanic\n\t/usr/local/go/src/runtime/panic.go:965\ngithub.com/milvus-io/milvus/internal/mq/msgstream.(*mqMsgStream).AsConsumerWithPosition\n\t/go/src/github.com/milvus-io/milvus/internal/mq/msgstream/mq_msgstream.go:174\ngithub.com/milvus-io/milvus/internal/mq/msgstream.(*mqMsgStream).AsConsumer\n\t/go/src/github.com/milvus-io/milvus/internal/mq/msgstream/mq_msgstream.go:131\ngithub.com/milvus-io/milvus/internal/mq/msgstream.UnsubscribeChannels\n\t/go/src/github.com/milvus-io/milvus/internal/mq/msgstream/msgstream_util.go:35\ngithub.com/milvus-io/milvus/internal/datacoord.(*ChannelManager).CleanupAndReassign\n\t/go/src/github.com/milvus-io/milvus/internal/datacoord/channel_manager.go:781\ngithub.com/milvus-io/milvus/internal/datacoord.(*ChannelManager).processAck\n\t/go/src/github.com/milvus-io/milvus/internal/datacoord/channel_manager.go:621\ngithub.com/milvus-io/milvus/internal/datacoord.(*ChannelManager).watchChannelStatesLoop\n\t/go/src/github.com/milvus-io/milvus/internal/datacoord/channel_manager.go:655"]
mdp-587-n-milvus-datacoord-5bd8cdb979-r2znf_milvus-ci_datacoord-c7c8715bf4447ee8467df2fdb3a5f4de1100c3610e13f30b45befd3dd952f646.log:[2022/06/11 19:19:44.568 +00:00] [FATAL] [logutil.go:134] [panic] [recover="\"Failed to create consumer by-dev-rootcoord-dml_48, error = All attempts results:\\nattempt #1:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #2:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #3:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #4:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #5:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #6:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #7:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #8:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #9:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #10:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #11:server error: ConsumerBusy: Exclusive consumer is already connected\\n\""] [stack="github.com/milvus-io/milvus/internal/util/logutil.LogPanic\n\t/go/src/github.com/milvus-io/milvus/internal/util/logutil/logutil.go:134\nruntime.gopanic\n\t/usr/local/go/src/runtime/panic.go:965\ngithub.com/milvus-io/milvus/internal/mq/msgstream.(*mqMsgStream).AsConsumerWithPosition\n\t/go/src/github.com/milvus-io/milvus/internal/mq/msgstream/mq_msgstream.go:174\ngithub.com/milvus-io/milvus/internal/mq/msgstream.(*mqMsgStream).AsConsumer\n\t/go/src/github.com/milvus-io/milvus/internal/mq/msgstream/mq_msgstream.go:131\ngithub.com/milvus-io/milvus/internal/mq/msgstream.UnsubscribeChannels\n\t/go/src/github.com/milvus-io/milvus/internal/mq/msgstream/msgstream_util.go:35\ngithub.com/milvus-io/milvus/internal/datacoord.(*ChannelManager).CleanupAndReassign\n\t/go/src/github.com/milvus-io/milvus/internal/datacoord/channel_manager.go:781\ngithub.com/milvus-io/milvus/internal/datacoord.(*ChannelManager).processAck\n\t/go/src/github.com/milvus-io/milvus/internal/datacoord/channel_manager.go:621\ngithub.com/milvus-io/milvus/internal/datacoord.(*ChannelManager).watchChannelStatesLoop\n\t/go/src/github.com/milvus-io/milvus/internal/datacoord/channel_manager.go:655"]
mdp-587-n-milvus-datacoord-5bd8cdb979-r2znf_milvus-ci_datacoord-ed84ab3e36ec35979c56297c08416ab76b1ec905f207745d9280bd814ae9905a.log:[2022/06/11 19:24:00.212 +00:00] [FATAL] [logutil.go:134] [panic] [recover="\"Failed to create consumer by-dev-rootcoord-dml_48, error = All attempts results:\\nattempt #1:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #2:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #3:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #4:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #5:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #6:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #7:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #8:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #9:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #10:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #11:server error: ConsumerBusy: Exclusive consumer is already connected\\n\""] [stack="github.com/milvus-io/milvus/internal/util/logutil.LogPanic\n\t/go/src/github.com/milvus-io/milvus/internal/util/logutil/logutil.go:134\nruntime.gopanic\n\t/usr/local/go/src/runtime/panic.go:965\ngithub.com/milvus-io/milvus/internal/mq/msgstream.(*mqMsgStream).AsConsumerWithPosition\n\t/go/src/github.com/milvus-io/milvus/internal/mq/msgstream/mq_msgstream.go:174\ngithub.com/milvus-io/milvus/internal/mq/msgstream.(*mqMsgStream).AsConsumer\n\t/go/src/github.com/milvus-io/milvus/internal/mq/msgstream/mq_msgstream.go:131\ngithub.com/milvus-io/milvus/internal/mq/msgstream.UnsubscribeChannels\n\t/go/src/github.com/milvus-io/milvus/internal/mq/msgstream/msgstream_util.go:35\ngithub.com/milvus-io/milvus/internal/datacoord.(*ChannelManager).CleanupAndReassign\n\t/go/src/github.com/milvus-io/milvus/internal/datacoord/channel_manager.go:781\ngithub.com/milvus-io/milvus/internal/datacoord.(*ChannelManager).processAck\n\t/go/src/github.com/milvus-io/milvus/internal/datacoord/channel_manager.go:621\ngithub.com/milvus-io/milvus/internal/datacoord.(*ChannelManager).watchChannelStatesLoop\n\t/go/src/github.com/milvus-io/milvus/internal/datacoord/channel_manager.go:655"]
mdp-587-n-milvus-datacoord-5bd8cdb979-r2znf_milvus-ci_datacoord-edd2f7f982bc8d362ac8b77ed5ebc8ad5859c206c018495a0b71bea50a0e1459.log:[2022/06/11 19:28:38.119 +00:00] [FATAL] [logutil.go:134] [panic] [recover="\"Failed to create consumer by-dev-rootcoord-dml_48, error = All attempts results:\\nattempt #1:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #2:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #3:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #4:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #5:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #6:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #7:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #8:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #9:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #10:server error: ConsumerBusy: Exclusive consumer is already connected\\nattempt #11:server error: ConsumerBusy: Exclusive consumer is already connected\\n\""] [stack="github.com/milvus-io/milvus/internal/util/logutil.LogPanic\n\t/go/src/github.com/milvus-io/milvus/internal/util/logutil/logutil.go:134\nruntime.gopanic\n\t/usr/local/go/src/runtime/panic.go:965\ngithub.com/milvus-io/milvus/internal/mq/msgstream.(*mqMsgStream).AsConsumerWithPosition\n\t/go/src/github.com/milvus-io/milvus/internal/mq/msgstream/mq_msgstream.go:174\ngithub.com/milvus-io/milvus/internal/mq/msgstream.(*mqMsgStream).AsConsumer\n\t/go/src/github.com/milvus-io/milvus/internal/mq/msgstream/mq_msgstream.go:131\ngithub.com/milvus-io/milvus/internal/mq/msgstream.UnsubscribeChannels\n\t/go/src/github.com/milvus-io/milvus/internal/mq/msgstream/msgstream_util.go:35\ngithub.com/milvus-io/milvus/internal/datacoord.(*ChannelManager).CleanupAndReassign\n\t/go/src/github.com/milvus-io/milvus/internal/datacoord/channel_manager.go:781\ngithub.com/milvus-io/milvus/internal/datacoord.(*ChannelManager).processAck\n\t/go/src/github.com/milvus-io/milvus/internal/datacoord/channel_manager.go:621\ngithub.com/milvus-io/milvus/internal/datacoord.(*ChannelManager).watchChannelStatesLoop\n\t/go/src/github.com/milvus-io/milvus/internal/datacoord/channel_manager.go:655"]
mdp-587-n-milvus-datanode-6d884f4486-5l8r2_milvus-ci_datanode-281cfb77c53141b19b662d6702857e5197c23faadca1befe60379cb3a03240d4.log:panic: All attempts results:
mdp-587-n-milvus-datanode-6d884f4486-9xbkr_milvus-ci_datanode-c60f0d712f1f9c8ddf1a6a7e3f6648accee02ee7f5574930307b3f7c37582694.log:panic: All attempts results:

Expected Behavior

No panic

Steps To Reproduce

Running nightly: [pipeline cluster pulsar]

https://ci.milvus.io:18080/jenkins/blue/organizations/jenkins/milvus-nightly-ci/detail/master/587/pipeline

Milvus Log

artifacts-milvus-distributed-pulsar-master-587-pymilvus-e2e-logs.tar.gz

Anything else?

No response