Fix kafka consumer double close (#25051)

Signed-off-by: NCongqi Xia <congqi.xia@zilliz.com>

pkg/mq/msgstream/mqwrapper/kafka/kafka_consumer.go

@@ -251,7 +251,6 @@ func (kc *Consumer) CheckTopicValid(topic string) error {
 func (kc *Consumer) Close() {
 	kc.closeOnce.Do(func() {
 		close(kc.closeCh)
-		kc.wg.Wait()
-		kc.c.Close()
+		kc.wg.Wait() // wait worker exist and close the client
 	})
 }