milvus: [Bug]: Kafka topics should be explicitly created with externally configured kafka

Is there an existing issue for this?

• I have searched the existing issues

Environment

- Milvus version: milvusdb/milvus-dev tag - 2.1.0
- Deployment mode(standalone or cluster): cluster
- SDK version(e.g. pymilvus v2.0.0rc2): go v2
- OS(Ubuntu or CentOS): ubuntu
- CPU/Memory: xx
- GPU: No
- Others: 

External kafka enabled. We are using managed kafka cluster through Confluent where we use a "Basic" or "Standard" cluster [1].

[1]

Current Behavior

With external kafka configured, Topics required for inter-server communication are not created, but are delegated to the kafka consumers which are configured with allow.auto.create.topics - https://github.com/milvus-io/milvus/blob/master/internal/mq/msgstream/mqwrapper/kafka/kafka_client.go#L121

Expected Behavior

With confluent managed kafka clusters, configuring behavior is currently not supported for the “Basic” and “Standard” clusters [1]. Also it seems that configuring this, may not result in intended behavior, see - https://github.com/confluentinc/confluent-kafka-go/issues/615

It would be great if these topics are created as part of the producer (this operation possibly could be made idempotent?) creation.

[1] https://docs.confluent.io/cloud/current/clusters/broker-config.html#ccloud-cluster-and-topic-configuration-settings

Steps To Reproduce

1. Setting up milvus cluster mode with the following changes:


"image": {
  "all": {
   "repository": "milvusdb/milvus-dev",
   "tag": "2.1.0-latest"
  }
},

"externalKafka": {
 "enabled": true,
 "brokerList": "brokerlist",
 "securityProtocol": "SASL_SSL",
 "sasl":
  "mechanisms": "PLAIN",
  "username": "username",
  "password": "password",
}

1. Deploying milvus on a kubernetes cluster
2. Using go sdk to create a collection using CreateCollection method. This fails with:

CreateCollection failed: send dd create collection req failed, error = Broker: Unknown topic or partition

1. In the logs of datacoord service/pod, we see:

[2022/06/30 23:58:54.152 +00:00] [ERROR] [kafka_consumer.go:143] ["consume msg failed"] [topic=by-dev-datacoord-timetick-channel] [groupID=by-dev-dataCoord] [error="by-dev-datacoord-timetick-channel [0]: topic does not exist (Broker: Unknown topic or partition)"] [stack="github.com/milvus-io/milvus/internal/mq/msgstream/mqwrapper/kafka.(*Consumer).Chan.func1.1\n\t/go/src/github.com/milvus-io/milvus/internal/mq/msgstream/mqwrapper/kafka/kafka_consumer.go:143"]

### Milvus Log

_No response_

### Anything else?

_No response_