add messageId and topic as props of DLQ message (#907)

Co-authored-by: Prabhudas Garule <[email protected]>

pulsar/consumer_test.go

@@ -1505,6 +1505,12 @@ func DLQWithProducerOptions(t *testing.T, prodOpt *ProducerOptions) {
 
 		expectMsg := fmt.Sprintf("hello-%d", expectedMsgIdx)
 		assert.Equal(t, []byte(expectMsg), msg.Payload())
+
+		// check original messageId
+		assert.NotEmpty(t, msg.Properties()[PropertyOriginMessageID])
+
+		// check original topic
+		assert.NotEmpty(t, msg.Properties()[SysPropertyRealTopic])
 	}
 
 	// No more messages on the DLQ

pulsar/dlq_router.go

@@ -92,11 +92,21 @@ func (r *dlqRouter) run() {
 			producer := r.getProducer(cm.Consumer.(*consumer).options.Schema)
 			msg := cm.Message.(*message)
 			msgID := msg.ID()
+
+			// properties associated with original message
+			properties := msg.Properties()
+
+			// include orinal message id in string format in properties
+			properties[PropertyOriginMessageID] = msgID.String()
+
+			// include original topic name of the message in properties
+			properties[SysPropertyRealTopic] = msg.Topic()
+
 			producer.SendAsync(context.Background(), &ProducerMessage{
 				Payload:             msg.Payload(),
 				Key:                 msg.Key(),
 				OrderingKey:         msg.OrderingKey(),
-				Properties:          msg.Properties(),
+				Properties:          properties,
 				EventTime:           msg.EventTime(),
 				ReplicationClusters: msg.replicationClusters,
 			}, func(MessageID, *ProducerMessage, error) {

pulsar/message.go

@@ -154,6 +154,9 @@ type MessageID interface {
 
 	// PartitionIdx returns the message partitionIdx
 	PartitionIdx() int32
+
+	// String returns message id in string format
+	String() string
 }
 
 // DeserializeMessageID reconstruct a MessageID object from its serialized representation

pulsar/reader_test.go

@@ -430,6 +430,14 @@ func (id *myMessageID) PartitionIdx() int32 {
 	return id.PartitionIdx()
 }
 
+func (id *myMessageID) String() string {
+	mid, err := DeserializeMessageID(id.data)
+	if err != nil {
+		return ""
+	}
+	return fmt.Sprintf("%d:%d:%d", mid.LedgerID(), mid.EntryID(), mid.PartitionIdx())
+}
+
 func TestReaderOnSpecificMessageWithCustomMessageID(t *testing.T) {
 	client, err := NewClient(ClientOptions{
 		URL: lookupURL,

pulsar/retry_router.go

@@ -35,6 +35,7 @@ const (
 	SysPropertyRetryTopic      = "RETRY_TOPIC"
 	SysPropertyReconsumeTimes  = "RECONSUMETIMES"
 	SysPropertyOriginMessageID = "ORIGIN_MESSAGE_IDY_TIME"
+	PropertyOriginMessageID    = "ORIGIN_MESSAGE_ID"
 )
 
 type RetryMessage struct {