[fix] [issue 1057]: Fix the producer flush opertion is not guarantee …

…to flush all messages (#1058)

Fixes #1057

### Motivation

`dataChan` is introduced by #1029  to fix the problem of reconnectToBroker. But it missed that if a flush operation excuted, there may still be some messages in `dataChan`. And these messages can't be flushed.

### Modifications

- Fix the producer flush opertion is not guarantee to flush all messages

(cherry picked from commit 9867c29)

pulsar/producer_partition.go

@@ -1048,6 +1048,15 @@ func (p *partitionProducer) internalFlushCurrentBatches() {
 }
 
 func (p *partitionProducer) internalFlush(fr *flushRequest) {
+	// clear all the messages which have sent to dataChan before flush
+	if len(p.dataChan) != 0 {
+		oldDataChan := p.dataChan
+		p.dataChan = make(chan *sendRequest, p.options.MaxPendingMessages)
+		for len(oldDataChan) != 0 {
+			pendingData := <-oldDataChan
+			p.internalSend(pendingData)
+		}
+	}
 
 	p.internalFlushCurrentBatch()