[Fix] DLQ Messages will lose system properties when sent from reconsumeLater() (#1392)

Fixes #1388
Master Issue: #907, apache/pulsar#23182

### Motivation
Since pulsar sdk does not support `AckTimeout()` method, we can only use `Nack()` or `ReconsumeLater()` to trigger dlq policy and send messages to dlq. We can find system properties(e.g. `REAL_TOPIC` and `ORIGIN_MESSAGE_ID`) in these dlq messages.

Refer to issue #1388, before pr #907 `dlq_router#run()` will directly copy properties to create a new dlq message when receiving a message from `dlq.Chan()`. But we only define message properties in ReconsumeLater's `ReconsumeLaterWithCustomProperties()` and not in Nack's `dispatcher()` before send to this `dlq.Chan()`, which lead to only dlq messages created by `ReconsumeLater()` can have these properties.

The above pr replaced this behavior with just adding system properties as soon as creating dlq messages in `dlq_router#run()`, which use `message.topic` as `REAL_TOPIC` property value. But messages sent from `ReconsumeLater() ReconsumeLaterWithCustomProperties()` do not contain related `message.topic` fields, in case it will override defined `REAL_TOPIC` to empty string.

### Modifications
- Add `REAL_TOPIC`, `ORIGIN_MESSAGE_IDY_TIME`, `ORIGIN_MESSAGE_ID` system properties both in ReconsumeLater's `ReconsumeLaterWithCustomProperties()` and Nack's `dispatcher()` functions.
- Add `Key`, `OrderingKey`, `EventTime` message fields both in `ReconsumeLater() ReconsumeLaterWithCustomProperties()` and `dlq_router run()` functions to keep consistent with [Java realization](apache/pulsar#23182), so that dlq messages can preserve source message info.
- Remove adding system properties strategy in `dlq_router#run()`.

Author: geniusjoe

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 {

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 {

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++

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) {

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