Fix incorrect ledgerID and entryID logged when producer receives unexpected ack

Author: RobertIndie

pulsar/producer_partition.go

@@ -1274,14 +1274,14 @@ func (p *partitionProducer) ReceivedSendReceipt(response *pb.CommandSendReceipt)
 
 	if pi.sequenceID < response.GetSequenceId() {
 		// Force connection closing so that messages can be re-transmitted in a new connection
-		p.log.Warnf("Received ack for %v on sequenceId %v - expected: %v, local < remote, closing connection",
-			response.GetMessageId(), response.GetSequenceId(), pi.sequenceID)
+		p.log.Warnf("Received ack for ledgerId:%d entryId:%d on sequenceId %v - expected: %v, local < remote, closing connection",
+			int64(*response.GetMessageId().LedgerId), int64(*response.GetMessageId().EntryId), response.GetSequenceId(), pi.sequenceID)
 		p._getConn().Close()
 		return
 	} else if pi.sequenceID > response.GetSequenceId() {
 		// Ignoring the ack since it's referring to a message that has already timed out.
-		p.log.Warnf("Received ack for %v on sequenceId %v - expected: %v, local > remote, ignore it",
-			response.GetMessageId(), response.GetSequenceId(), pi.sequenceID)
+		p.log.Warnf("Received ack for ledgerId:%d entryId:%d on sequenceId %v - expected: %v, local > remote, ignore it",
+			int64(*response.GetMessageId().LedgerId), int64(*response.GetMessageId().EntryId), response.GetSequenceId(), pi.sequenceID)
 		return
 	} else {
 		// The ack was indeed for the expected item in the queue, we can remove it and trigger the callback

pulsar/producer_test.go

@@ -2357,6 +2357,36 @@ func TestFailPendingMessageWithClose(t *testing.T) {
 	assert.Equal(t, 0, partitionProducerImp.pendingQueue.Size())
 }
 
+func TestProducerSendDuplicatedMessages(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+	assert.NoError(t, err)
+	defer client.Close()
+	testProducer, err := client.CreateProducer(ProducerOptions{
+		Topic: newTopicName(),
+	})
+
+	assert.NoError(t, err)
+	assert.NotNil(t, testProducer)
+	_, err = testProducer.Send(context.Background(), &ProducerMessage{
+		Payload: make([]byte, 1024),
+	})
+	assert.NoError(t, err)
+	for i := 0; i < 3; i++ {
+		var seqId int64 = 0
+		msgId, err := testProducer.Send(context.Background(), &ProducerMessage{
+			Payload:    make([]byte, 1024),
+			SequenceID: &seqId,
+		})
+		assert.NoError(t, err)
+		assert.NotNil(t, msgId)
+		assert.Equal(t, int64(-1), msgId.LedgerID())
+		assert.Equal(t, int64(-1), msgId.EntryID())
+	}
+	testProducer.Close()
+}
+
 type pendingQueueWrapper struct {
 	pendingQueue   internal.BlockingQueue
 	writtenBuffers *[]internal.Buffer