[Fix][producer] Ensure all data in dataChan will be processed when in…

…ternalClose() was called

pulsar/producer_partition.go

@@ -1266,11 +1266,27 @@ func (p *partitionProducer) ReceivedSendReceipt(response *pb.CommandSendReceipt)
 
 func (p *partitionProducer) internalClose(req *closeProducer) {
 	defer close(req.doneCh)
+	defer close(p.dataChan)
+	defer close(p.cmdChan)
 	if !p.casProducerState(producerReady, producerClosing) {
 		return
 	}
 
 	p.log.Info("Closing producer")
+	for {
+		if len(p.dataChan) == 0 {
+			break
+		}
+		data := <-p.dataChan
+		p.internalSend(data)
+	}
+	flushReq := &flushRequest{
+		doneCh: make(chan struct{}),
+		err:    nil,
+	}
+	p.internalFlush(flushReq)
+	// wait for the flush request to complete
+	<-flushReq.doneCh
 
 	id := p.client.rpcClient.NewRequestID()
 	_, err := p.client.rpcClient.RequestOnCnx(p._getConn(), id, pb.BaseCommand_CLOSE_PRODUCER, &pb.CommandCloseProducer{