From ffbb581362df67edd83c21917617fc5e86e44807 Mon Sep 17 00:00:00 2001
From: horizonzy <horizonzy@apache.org>
Date: Wed, 5 Feb 2025 10:37:42 +0800
Subject: [PATCH] Fix get properties failed when the message metadata contains
 the same key, but with different values issue.

---
 .../pulsar/common/api/raw/RawMessageImpl.java |  3 ++-
 .../common/api/raw/RawMessageImplTest.java    | 20 ++++++++++++++++++-
 2 files changed, 21 insertions(+), 2 deletions(-)

diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/api/raw/RawMessageImpl.java b/pulsar-common/src/main/java/org/apache/pulsar/common/api/raw/RawMessageImpl.java
index daf7fd6a7644c..8240f234aa1ed 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/api/raw/RawMessageImpl.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/api/raw/RawMessageImpl.java
@@ -99,7 +99,8 @@ public Map<String, String> getProperties() {
                               (oldValue, newValue) -> newValue));
         } else if (msgMetadata.getMetadata().getPropertiesCount() > 0) {
             return msgMetadata.getMetadata().getPropertiesList().stream()
-                    .collect(Collectors.toMap(KeyValue::getKey, KeyValue::getValue));
+                    .collect(Collectors.toMap(KeyValue::getKey, KeyValue::getValue,
+                            (oldValue, newValue) -> newValue));
         } else {
             return Collections.emptyMap();
         }
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/api/raw/RawMessageImplTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/api/raw/RawMessageImplTest.java
index b0b9d42866cba..61783b36dca22 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/api/raw/RawMessageImplTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/api/raw/RawMessageImplTest.java
@@ -41,7 +41,7 @@ public class RawMessageImplTest {
     private static final String HARD_CODE_KEY_ID_VALUE  = "__pfn_input_msg_id_value__";
 
     @Test
-    public void testGetProperties() {
+    public void testGetMessageSingleMetadataProperties() {
         ReferenceCountedMessageMetadata refCntMsgMetadata =
                 ReferenceCountedMessageMetadata.get(mock(ByteBuf.class));
         SingleMessageMetadata singleMessageMetadata = new SingleMessageMetadata();
@@ -56,6 +56,24 @@ public void testGetProperties() {
         assertEquals(HARD_CODE_KEY_ID_VALUE, properties.get(HARD_CODE_KEY_ID));
     }
 
+    @Test
+    public void testGetMessageMetadataProperties() {
+        ReferenceCountedMessageMetadata refCntMsgMetadata =
+                ReferenceCountedMessageMetadata.get(mock(ByteBuf.class));
+
+        MessageMetadata messageMetadata = refCntMsgMetadata.getMetadata();
+        messageMetadata.addProperty().setKey(HARD_CODE_KEY).setValue(KEY_VALUE_FIRST);
+        messageMetadata.addProperty().setKey(HARD_CODE_KEY).setValue(KEY_VALUE_SECOND);
+        messageMetadata.addProperty().setKey(HARD_CODE_KEY_ID).setValue(HARD_CODE_KEY_ID_VALUE);
+
+        RawMessage msg = RawMessageImpl.get(refCntMsgMetadata, null, null, 0, 0, 0);
+        Map<String, String> properties = msg.getProperties();
+        assertEquals(properties.get(HARD_CODE_KEY), KEY_VALUE_SECOND);
+        assertEquals(properties.get(HARD_CODE_KEY_ID), HARD_CODE_KEY_ID_VALUE);
+        assertEquals(KEY_VALUE_SECOND, properties.get(HARD_CODE_KEY));
+        assertEquals(HARD_CODE_KEY_ID_VALUE, properties.get(HARD_CODE_KEY_ID));
+    }
+
     @Test
     public void testNonBatchedMessage() {
         MessageMetadata messageMetadata = new MessageMetadata();