Repository: apache/pulsar-client-go

Issue:
Missing REAL_TOPIC in DLQ message when max deliveries exceeeds
#### Expected behavior

In DLQ message, the `REAL_TOPIC` should be present.

#### Actual behavior

Now it is empty.

#### Steps to reproduce

Configure for example the MaxDeliveries to 1.
Consume and call ReconsumeAfter twice.
The message present in DLQ does have REAL_TOPIC property but it is empty.

#### System configuration
**Pulsar version**: v3.6.4
**Pulsar-client-go version**: > v0.14.0 (tested version, historical tags were not tested)

#### Additional Info

After looking at the code of `ReconsumeLaterWithCustomProperties`
I've found out that the constructed message is not affected with the `topic` props:
```
	reconsumeTimes := 1
	if s, ok := props[SysPropertyReconsumeTimes]; ok {
		reconsumeTimes, _ = strconv.Atoi(s)
		reconsumeTimes++
	} else {
		props[SysPropertyRealTopic] = msg.Topic()
		props[SysPropertyOriginMessageID] = msgID.messageID.String()
	}
	props[SysPropertyReconsumeTimes] = strconv.Itoa(reconsumeTimes)
	props[SysPropertyDelayTime] = fmt.Sprintf("%d", int64(delay)/1e6)

	consumerMsg := ConsumerMessage{
		Consumer: c,
		Message: &message{
			payLoad:    msg.Payload(),
			properties: props,
			msgID:      msgID,
		},
	}

```

That prop is used inside `dlq_router.run`:
```
			// include original topic name of the message in properties
			properties[SysPropertyRealTopic] = msg.Topic()
```

That would explain the blank `REAL_TOPIC` value in dlq message.


Hints from issue comments:
@YanshuoH: Looking at version v0.9.0, there is no real topic override in the `dlq_router.run`

```
func (r *dlqRouter) run() {
	for {
		select {
		case cm := <-r.messageCh:
			r.log.WithField("msgID", cm.ID()).Debug("Got message for DLQ")
			producer := r.getProducer(cm.Consumer.(*consumer).options.Schema)
			msg := cm.Message.(*message)
			msgID := msg.ID()
			producer.SendAsync(context.Background(), &ProducerMessage{
				Payload:             msg.Payload(),
				Key:                 msg.Key(),
				OrderingKey:         msg.OrderingKey(),
				Properties:          msg.Properties(),
				EventTime:           msg.EventTime(),
				ReplicationClusters: msg.replicationClusters,
			}, func(MessageID, *ProducerMessage, error) {
				r.log.WithField("msgID", msgID).Debug("Sent message to DLQ")

				// The Producer ack might be coming from the connection go-routine that
				// is also used by the consumer. In that case we would get a dead-lock
				// if we'd try to ack.
				go cm.Consumer.AckID(msgID)
			})

		case <-r.closeCh:
			if r.producer != nil {
				r.producer.Close()
			}
			r.log.Debug("Closed DLQ router")
			return
		}
	}
}
```

I think it comes from this PR: https://github.com/apache/pulsar-client-go/pull/907/files

Solution patch:
diff --git a/pulsar/consumer_impl.go b/pulsar/consumer_impl.go
index c8b28b1..5f6dac3 100644
--- a/pulsar/consumer_impl.go
+++ b/pulsar/consumer_impl.go
@@ -632,17 +632,21 @@ func (c *consumer) ReconsumeLaterWithCustomProperties(msg Message, customPropert
 	} else {
 		props[SysPropertyRealTopic] = msg.Topic()
 		props[SysPropertyOriginMessageID] = msgID.messageID.String()
+		props[PropertyOriginMessageID] = msgID.messageID.String()
 	}
 	props[SysPropertyReconsumeTimes] = strconv.Itoa(reconsumeTimes)
 	props[SysPropertyDelayTime] = fmt.Sprintf("%d", int64(delay)/1e6)
 
 	consumerMsg := ConsumerMessage{
 		Consumer: c,
+		// Copy msgID so that dlq/rlq router can ack this msg after successfully sent to new topic
 		Message: &message{
-			payLoad:    msg.Payload(),
-			properties: props,
-			msgID:      msgID,
-			eventTime:  msg.EventTime(),
+			payLoad:     msg.Payload(),
+			key:         msg.Key(),
+			orderingKey: msg.OrderingKey(),
+			properties:  props,
+			eventTime:   msg.EventTime(),
+			msgID:       msgID,
 		},
 	}
 	if uint32(reconsumeTimes) > c.dlq.policy.MaxDeliveries {
diff --git a/pulsar/consumer_partition.go b/pulsar/consumer_partition.go
index 2a2f4eb..41341ed 100644
--- a/pulsar/consumer_partition.go
+++ b/pulsar/consumer_partition.go
@@ -1606,6 +1606,26 @@ func (pc *partitionConsumer) dispatcher() {
 			if !pc.isSeeking.Load() {
 				if pc.dlq.shouldSendToDlq(&nextMessage) {
 					// pass the message to the DLQ router
+					// we need to create a new ConsumerMessage and add dlq related metadata properties
+					properties := make(map[string]string)
+					properties[SysPropertyRealTopic] = messages[0].Topic()
+					properties[SysPropertyOriginMessageID] = messages[0].msgID.String()
+					properties[PropertyOriginMessageID] = messages[0].msgID.String()
+					for key, value := range messages[0].properties {
+						properties[key] = value
+					}
+					nextMessage = ConsumerMessage{
+						Consumer: pc.parentConsumer,
+						// Copy msgID so that dlq/rlq router can ack this msg after successfully sent to new topic
+						Message: &message{
+							payLoad:     messages[0].Payload(),
+							key:         messages[0].Key(),
+							orderingKey: messages[0].OrderingKey(),
+							properties:  properties,
+							eventTime:   messages[0].EventTime(),
+							msgID:       messages[0].msgID,
+						},
+					}
 					pc.metrics.DlqCounter.Inc()
 					messageCh = pc.dlq.Chan()
 				} else {
diff --git a/pulsar/dlq_router.go b/pulsar/dlq_router.go
index c65f01a..3fef99a 100644
--- a/pulsar/dlq_router.go
+++ b/pulsar/dlq_router.go
@@ -111,21 +111,11 @@ 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:          properties,
+				Properties:          msg.Properties(),
 				EventTime:           msg.EventTime(),
 				ReplicationClusters: msg.replicationClusters,
 			}, func(_ MessageID, _ *ProducerMessage, err error) {
diff --git a/pulsar/message.go b/pulsar/message.go
index f965555..2cc0472 100644
--- a/pulsar/message.go
+++ b/pulsar/message.go
@@ -40,7 +40,7 @@ type ProducerMessage struct {
 
 	// EventTime set the event time for a given message
 	// By default, messages don't have an event time associated, while the publish
-	// time will be be always present.
+	// time will be always present.
 	// Set the event time to a non-zero timestamp to explicitly declare the time
 	// that the event "happened", as opposed to when the message is being published.
 	EventTime time.Time



Test patch:
diff --git a/pulsar/consumer_test.go b/pulsar/consumer_test.go
index 4f67c36..c151c98 100644
--- a/pulsar/consumer_test.go
+++ b/pulsar/consumer_test.go
@@ -1580,12 +1580,26 @@ func DLQWithProducerOptions(t *testing.T, prodOpt *ProducerOptions) {
 	defer producer.Close()
 
 	// send 10 messages
+	eventTimeList := make([]time.Time, 10)
+	msgIDList := make([]string, 10)
+	msgKeyList := make([]string, 10)
 	for i := 0; i < 10; i++ {
-		if _, err := producer.Send(ctx, &ProducerMessage{
-			Payload: []byte(fmt.Sprintf("hello-%d", i)),
-		}); err != nil {
+		eventTime := time.Now()
+		eventTimeList[i] = eventTime
+		msgKeyList[i] = fmt.Sprintf("key-%d", i)
+		msgID, err := producer.Send(ctx, &ProducerMessage{
+			Payload:     []byte(fmt.Sprintf("hello-%d", i)),
+			Key:         fmt.Sprintf("key-%d", i),
+			OrderingKey: fmt.Sprintf("key-%d", i),
+			EventTime:   eventTime,
+			Properties: map[string]string{
+				"key": fmt.Sprintf("key-%d", i),
+			},
+		})
+		if err != nil {
 			log.Fatal(err)
 		}
+		msgIDList[i] = msgID.String()
 	}
 
 	// receive 10 messages and only ack half-of-them
@@ -1624,10 +1638,27 @@ func DLQWithProducerOptions(t *testing.T, prodOpt *ProducerOptions) {
 		assert.True(t, regex.MatchString(msg.ProducerName()))
 
 		// check original messageId
+		assert.NotEmpty(t, msg.Properties()[SysPropertyOriginMessageID])
+		assert.Equal(t, msgIDList[expectedMsgIdx], msg.Properties()[SysPropertyOriginMessageID])
 		assert.NotEmpty(t, msg.Properties()[PropertyOriginMessageID])
+		assert.Equal(t, msgIDList[expectedMsgIdx], msg.Properties()[PropertyOriginMessageID])
 
 		// check original topic
-		assert.NotEmpty(t, msg.Properties()[SysPropertyRealTopic])
+		assert.Contains(t, msg.Properties()[SysPropertyRealTopic], topic)
+
+		// check original key
+		assert.NotEmpty(t, msg.Key())
+		assert.Equal(t, msgKeyList[expectedMsgIdx], msg.Key())
+		assert.NotEmpty(t, msg.OrderingKey())
+		assert.Equal(t, msgKeyList[expectedMsgIdx], msg.OrderingKey())
+		assert.NotEmpty(t, msg.Properties()["key"])
+		assert.Equal(t, msg.Key(), msg.Properties()["key"])
+
+		//	check original event time
+		//	Broker will ignore event time microsecond(us) level precision,
+		//	so that we need to check eventTime precision in millisecond level
+		assert.NotEqual(t, 0, msg.EventTime())
+		assert.True(t, eventTimeList[expectedMsgIdx].Sub(msg.EventTime()).Abs() < 2*time.Millisecond)
 	}
 
 	// No more messages on the DLQ
@@ -1855,9 +1886,24 @@ func TestRLQ(t *testing.T) {
 	assert.Nil(t, err)
 	defer producer.Close()
 
+	eventTimeList := make([]time.Time, N)
+	msgIDList := make([]string, N)
+	msgKeyList := make([]string, N)
 	for i := 0; i < N; i++ {
-		_, err = producer.Send(ctx, &ProducerMessage{Payload: []byte(fmt.Sprintf("MESSAGE_%d", i))})
+		eventTime := time.Now()
+		eventTimeList[i] = eventTime
+		msgKeyList[i] = fmt.Sprintf("key-%d", i)
+		msgID, err := producer.Send(ctx, &ProducerMessage{
+			Payload:     []byte(fmt.Sprintf("MESSAGE_%d", i)),
+			Key:         fmt.Sprintf("key-%d", i),
+			OrderingKey: fmt.Sprintf("key-%d", i),
+			EventTime:   eventTime,
+			Properties: map[string]string{
+				"key": fmt.Sprintf("key-%d", i),
+			},
+		})
 		assert.Nil(t, err)
+		msgIDList[i] = msgID.String()
 	}
 
 	// 2. Create consumer on the Retry Topic to reconsume N messages (maxRedeliveries+1) times
@@ -1903,6 +1949,33 @@ func TestRLQ(t *testing.T) {
 	dlqReceived := 0
 	for dlqReceived < N {
 		msg, err := dlqConsumer.Receive(ctx)
+		//	check original messageId
+		//	we create a topic with three partitions,
+		//	so that messages maybe not be received as the same order as we produced
+		assert.NotEmpty(t, msg.Properties()[SysPropertyOriginMessageID])
+		assert.Contains(t, msgIDList, msg.Properties()[SysPropertyOriginMessageID])
+		assert.NotEmpty(t, msg.Properties()[PropertyOriginMessageID])
+		assert.Contains(t, msgIDList, msg.Properties()[PropertyOriginMessageID])
+
+		// check original topic
+		assert.Contains(t, msg.Properties()[SysPropertyRealTopic], topic)
+
+		// check original key
+		assert.NotEmpty(t, msg.Key())
+		assert.Contains(t, msgKeyList, msg.Key())
+		assert.NotEmpty(t, msg.OrderingKey())
+		assert.Contains(t, msgKeyList, msg.OrderingKey())
+		assert.NotEmpty(t, msg.Properties()["key"])
+		assert.Equal(t, msg.Key(), msg.Properties()["key"])
+
+		// check original event time
+		assert.NotEqual(t, 0, msg.EventTime())
+		//	check original event time
+		//	Broker will ignore event time microsecond(us) level precision,
+		//	so that we need to check eventTime precision in millisecond level
+		assert.LessOrEqual(t, eventTimeList[0].Add(-2*time.Millisecond), msg.EventTime())
+		assert.LessOrEqual(t, msg.EventTime(), eventTimeList[N-1].Add(2*time.Millisecond))
+
 		assert.Nil(t, err)
 		dlqConsumer.Ack(msg)
 		dlqReceived++