Close kafka internal consumer&producer properly (#24997)

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

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

@@ -22,6 +22,7 @@ type Consumer struct {
 	chanOnce   sync.Once
 	closeOnce  sync.Once
 	closeCh    chan struct{}
+	wg         sync.WaitGroup
 }
 
 const timeout = 3000
@@ -116,7 +117,9 @@ func (kc *Consumer) Chan() <-chan mqwrapper.Message {
 		panic("failed to chan a kafka consumer without assign")
 	}
 	kc.chanOnce.Do(func() {
+		kc.wg.Add(1)
 		go func() {
+			defer kc.wg.Done()
 			for {
 				select {
 				case <-kc.closeCh:
@@ -222,5 +225,7 @@ func (kc *Consumer) GetLatestMsgID() (mqwrapper.MessageID, error) {
 func (kc *Consumer) Close() {
 	kc.closeOnce.Do(func() {
 		close(kc.closeCh)
+		kc.wg.Wait() // wait work goroutine exit
+		kc.c.Close()
 	})
 }

internal/mq/msgstream/mqwrapper/kafka/kafka_producer.go

@@ -89,5 +89,7 @@ func (kp *kafkaProducer) Close() {
 		if cost > 500 {
 			log.Info("kafka producer is closed", zap.Any("topic", kp.topic), zap.Int64("time cost(ms)", cost))
 		}
+
+		kp.p.Close()
 	})
 }