[pulsar-client-go] branch master updated: Fixed batching flag logic (#209)

[mm...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

mmerli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar-client-go.git


The following commit(s) were added to refs/heads/master by this push:
     new d27cd74  Fixed batching flag logic (#209)
d27cd74 is described below

commit d27cd74fb7a3bc049af1f1c75534cfc45b240825
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Mon Apr 6 10:12:34 2020 -0700

    Fixed batching flag logic (#209)
    
    * Fixed batching flag logic
    
    * Fixed naming
---
 pulsar/producer_partition.go |  2 +-
 pulsar/producer_test.go      | 15 ++++++++++++++-
 2 files changed, 15 insertions(+), 2 deletions(-)

diff --git a/pulsar/producer_partition.go b/pulsar/producer_partition.go
index 58f512f..1241d47 100644
--- a/pulsar/producer_partition.go
+++ b/pulsar/producer_partition.go
@@ -241,7 +241,7 @@ func (p *partitionProducer) internalSend(request *sendRequest) {
 
 	sendAsBatch := !p.options.DisableBatching &&
 		msg.ReplicationClusters == nil &&
-		deliverAt.UnixNano() == 0
+		deliverAt.UnixNano() < 0
 
 	smm := &pb.SingleMessageMetadata{
 		PayloadSize: proto.Int(len(msg.Payload)),
diff --git a/pulsar/producer_test.go b/pulsar/producer_test.go
index 50eea77..a45d8ce 100644
--- a/pulsar/producer_test.go
+++ b/pulsar/producer_test.go
@@ -313,10 +313,23 @@ func TestFlushInProducer(t *testing.T) {
 	assert.Nil(t, err)
 	wg.Wait()
 
+	var ledgerID int64 = -1
+	var entryID int64 = -1
+
 	for i := 0; i < numOfMessages/2; i++ {
-		_, err = consumer.Receive(ctx)
+		msg, err := consumer.Receive(ctx)
 		assert.Nil(t, err)
 		msgCount++
+
+		msgID := msg.ID().(*messageID)
+		// Since messages are batched, they will be sharing the same ledgerId/entryId
+		if ledgerID == -1 {
+			ledgerID = msgID.ledgerID
+			entryID = msgID.entryID
+		} else {
+			assert.Equal(t, ledgerID, msgID.ledgerID)
+			assert.Equal(t, entryID, msgID.entryID)
+		}
 	}
 
 	assert.Equal(t, msgCount, numOfMessages/2)