fix: batch flush method (#476)

Signed-off-by: jonyhy96 <hy352144278@gmail.com>

Fixes #475

### Motivation

Fix wrong batch flush method bug.

### Modifications

Check if batchBuilder is multi batches before flush.

pulsar/producer_partition.go

@@ -397,7 +397,11 @@ func (p *partitionProducer) internalSend(request *sendRequest) {
 		msg.ReplicationClusters, deliverAt)
 	if !added {
 		// The current batch is full.. flush it and retry
-		p.internalFlushCurrentBatch()
+		if p.batchBuilder.IsMultiBatches() {
+			p.internalFlushCurrentBatches()
+		} else {
+			p.internalFlushCurrentBatch()
+		}
 
 		// after flushing try again to add the current payload
 		if ok := p.batchBuilder.Add(smm, p.sequenceIDGenerator, payload, request,