Fixed producer.Flush() (#5)

billowqiu · May 15, 2019 · d6a6187 · d6a6187
1 parent 9e3736a
commit d6a6187
Showing 1 changed file with 6 additions and 2 deletions.
diff --git a/pulsar/impl_partition_producer.go b/pulsar/impl_partition_producer.go
@@ -281,6 +281,7 @@ func (p *partitionProducer) internalFlush(fr *flushRequest) {
 
 	pi := p.pendingQueue.PeekLast().(*pendingItem)
 	pi.sendRequests = append(pi.sendRequests, &sendRequest{
+		msg: nil,
 		callback: func(id MessageID, message *ProducerMessage, e error) {
 			fr.err = e
 			fr.waitGroup.Done()
@@ -337,9 +338,12 @@ func (p *partitionProducer) ReceivedSendReceipt(response *pb.CommandSendReceipt)
 	p.pendingQueue.Poll()
 	for idx, i := range pi.sendRequests {
 		sr := i.(*sendRequest)
-		atomic.StoreInt64(&p.lastSequenceID, int64(pi.sequenceId))
-		if sr.callback != nil {
+		if sr.msg != nil {
+			atomic.StoreInt64(&p.lastSequenceID, int64(pi.sequenceId))
 			p.publishSemaphore.Release()
+		}
+
+		if sr.callback != nil {
 			msgID := newMessageId(
 				int64(response.MessageId.GetLedgerId()),
 				int64(response.MessageId.GetEntryId()),