[2.x] Support getting producerName and topic name from the received message (#332)

* Fix single topic consumer returning message ID with empty topic name (#330)

(cherry picked from commit fc04d7b)

* Add producer name attribute to the Message (#331)

* Add producer name attribute to the Message

* Fix compilation issue

(cherry picked from commit 8738e79)

* Remove unnesessary changes

Author: RobertIndie

src/Pulsar.Client/Common/DTO.fs

@@ -213,6 +213,7 @@ type internal Metadata =
         EncryptionAlgo: string
         OrderingKey: byte[]
         ReplicatedFrom: string
+        ProducerName: string
     }
 
 type MessageKey =
@@ -264,7 +265,7 @@ type Message<'T> internal (messageId: MessageId, data: byte[], key: PartitionKey
                   properties: IReadOnlyDictionary<string, string>, encryptionCtx: EncryptionContext option,
                   schemaVersion: byte[], sequenceId: SequenceId, orderingKey: byte[], publishTime: TimeStamp,
                   eventTime: Nullable<TimeStamp>,
-                  redeliveryCount: int32, replicatedFrom: string,
+                  redeliveryCount: int32, replicatedFrom: string, producerName: string,
                   getValue: unit -> 'T) =
     /// Get the unique message ID associated with this message.
     member this.MessageId = messageId
@@ -293,26 +294,27 @@ type Message<'T> internal (messageId: MessageId, data: byte[], key: PartitionKey
     member this.RedeliveryCount = redeliveryCount
     /// Get name of cluster, from which the message is replicated.
     member this.ReplicatedFrom = replicatedFrom
-    
+    /// Get name of producer of the message
+    member this.ProducerName = producerName
     /// Get the de-serialized value of the message, according the configured Schema.
     member this.GetValue() =
         getValue()
 
     member internal this.WithMessageId messageId =
         Message(messageId, data, key, hasBase64EncodedKey, properties, encryptionCtx, schemaVersion, sequenceId,
-                orderingKey, publishTime, eventTime, redeliveryCount, replicatedFrom, getValue)
+                orderingKey, publishTime, eventTime, redeliveryCount, replicatedFrom, producerName, getValue)
     /// Get a new instance of the message with updated data
     member this.WithData data =
         Message(messageId, data, key, hasBase64EncodedKey, properties, encryptionCtx, schemaVersion, sequenceId,
-                orderingKey, publishTime, eventTime, redeliveryCount, replicatedFrom, getValue)
+                orderingKey, publishTime, eventTime, redeliveryCount, replicatedFrom, producerName, getValue)
     /// Get a new instance of the message with updated key
     member this.WithKey (key, hasBase64EncodedKey) =
         Message(messageId, data, key, hasBase64EncodedKey, properties, encryptionCtx, schemaVersion, sequenceId,
-                orderingKey, publishTime, eventTime, redeliveryCount, replicatedFrom, getValue)
+                orderingKey, publishTime, eventTime, redeliveryCount, replicatedFrom, producerName, getValue)
     /// Get a new instance of the message with updated properties
     member this.WithProperties properties =
         Message(messageId, data, key, hasBase64EncodedKey, properties, encryptionCtx, schemaVersion, sequenceId,
-                orderingKey, publishTime, eventTime, redeliveryCount, replicatedFrom, getValue)
+                orderingKey, publishTime, eventTime, redeliveryCount, replicatedFrom, producerName, getValue)
 
 type Messages<'T> internal(maxNumberOfMessages: int, maxSizeOfMessages: int64) =
 

src/Pulsar.Client/Internal/ClientCnx.fs

@@ -526,6 +526,7 @@ and internal ClientCnx (config: PulsarClientConfiguration,
             EncryptionAlgo = messageMetadata.EncryptionAlgo
             OrderingKey = messageMetadata.OrderingKey
             ReplicatedFrom = messageMetadata.ReplicatedFrom
+            ProducerName = messageMetadata.ProducerName
         }
 
         {

src/Pulsar.Client/Internal/ConsumerImpl.fs

@@ -452,7 +452,7 @@ type internal ConsumerImpl<'T> (consumerConfig: ConsumerConfiguration<'T>, clien
         ackRequests.Clear()
 
     let getNewIndividualMsgIdWithPartition messageId =
-        { messageId with Type = MessageIdType.Single; Partition = partitionIndex; TopicName = %"" }
+        { messageId with Type = MessageIdType.Single; Partition = partitionIndex; TopicName = topicName.CompleteTopicName }
 
     let processPossibleToDLQ (messageId : MessageId) =
         let acknowledge = trySendAcknowledge Individual EmptyProperties None
@@ -685,6 +685,7 @@ type internal ConsumerImpl<'T> (consumerConfig: ConsumerConfiguration<'T>, clien
                             rawMessage.Metadata.EventTime,
                             rawMessage.RedeliveryCount,
                             rawMessage.Metadata.ReplicatedFrom,
+                            rawMessage.Metadata.ProducerName,
                             getValue
                         )
             if (rawMessage.RedeliveryCount >= deadLettersProcessor.MaxRedeliveryCount) then
@@ -1336,13 +1337,12 @@ type internal ConsumerImpl<'T> (consumerConfig: ConsumerConfiguration<'T>, clien
             elif rawMessage.AckSet.Count > 0 && not rawMessage.AckSet[i] then
                 skippedMessages <- skippedMessages + 1
             else
-                let messageId =
-                    {
-                        rawMessage.MessageId with
-                            Partition = partitionIndex
-                            Type = Batch(%i, acker)
-                            TopicName = %""
-                    }
+                let messageId = {
+                    rawMessage.MessageId with
+                        Partition = partitionIndex
+                        Type = Batch(%i, acker)
+                        TopicName = topicName.CompleteTopicName
+                }
                 let msgKey = singleMessageMetadata.PartitionKey
                 let getValue () =
                     keyValueProcessor
@@ -1374,6 +1374,7 @@ type internal ConsumerImpl<'T> (consumerConfig: ConsumerConfiguration<'T>, clien
                                 eventTime,
                                 rawMessage.RedeliveryCount,
                                 rawMessage.Metadata.ReplicatedFrom,
+                                rawMessage.Metadata.ProducerName,
                                 getValue
                             )
                 if (rawMessage.RedeliveryCount >= deadLettersProcessor.MaxRedeliveryCount) then

tests/IntegrationTests/Basic.fs

@@ -22,20 +22,22 @@ let tests =
 
     testList "Basic" [
 
-        testTask "Sent messageId should be equal to received messageId" {
+        testTask "Sent message/messageId should be equal to received message/messageId" {
 
             Log.Debug("Started Sent messageId should be equal to received messageId")
             let client = getClient()
-            let topicName = "public/default/topic-" + Guid.NewGuid().ToString("N")
+            let topicName = "persistent://public/default/topic-" + Guid.NewGuid().ToString("N")
 
             let! (producer1 : IProducer<byte[]>) =
                 client.NewProducer()
                     .Topic(topicName)
+                    .ProducerName("producer1")
                     .CreateAsync()
 
             let! (producer2 : IProducer<byte[]>) =
                 client.NewProducer()
                     .Topic(topicName)
+                    .ProducerName("producer2")
                     .EnableBatching(false)
                     .CreateAsync()
 
@@ -52,9 +54,13 @@ let tests =
 
             Expect.isTrue "" (msg1Id = msg1.MessageId)
             Expect.equal "" [| 0uy |] <| msg1.GetValue()
+            Expect.equal "Message ID topic name should match" topicName (string msg1.MessageId.TopicName)
+            Expect.equal "Message producer name should match" "producer1" (string msg1.ProducerName)
 
             Expect.isTrue "" (msg2Id = msg2.MessageId)
             Expect.equal "" [| 1uy |] <| msg2.GetValue()
+            Expect.equal "Message ID topic name should match" topicName (string msg2.MessageId.TopicName)
+            Expect.equal "Message producer name should match" "producer2" (string msg2.ProducerName)
 
             Log.Debug("Finished Sent messageId should be equal to received messageId")
         }

tests/UnitTests/Internal/ChunkedMessageTrackerTests.fs

@@ -30,6 +30,7 @@ let tests =
             EventTime = Nullable()
             OrderingKey = [||]
             ReplicatedFrom = ""
+            ProducerName = ""
         }
 
     let testRawMessage =