You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2024/10/21 13:34:14 UTC
(pulsar) 04/05: [improve][test] Added message properties tests for batch and non-batch messages (#23473)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 986a4db22f46d1bbeb878a7133863b192157ec59
Author: Penghui Li <pe...@apache.org>
AuthorDate: Fri Oct 18 19:31:48 2024 +0800
[improve][test] Added message properties tests for batch and non-batch messages (#23473)
(cherry picked from commit 8de27a2116596ce63d7cd09534b168dbb64f64da)
---
.../pulsar/broker/admin/PersistentTopicsTest.java | 72 ++++++++++++++++++++++
1 file changed, 72 insertions(+)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
index 65b821d6ddc..f453d573acd 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
@@ -1795,4 +1795,76 @@ public class PersistentTopicsTest extends MockedPulsarServiceBaseTest {
assertEquals(cursor.getCursorProperties().size(), 1);
assertEquals(cursor.getCursorProperties().get("foo"), "bar");
}
+
+ @Test
+ public void testPeekMessageWithProperties() throws Exception {
+ String topicName = "persistent://" + testTenant + "/" + testNamespaceLocal + "/testPeekMessageWithProperties";
+ admin.topics().createNonPartitionedTopic(topicName);
+
+ // Test non-batch messages
+ @Cleanup
+ Producer<String> nonBatchProducer = pulsarClient.newProducer(Schema.STRING)
+ .topic(topicName)
+ .enableBatching(false)
+ .create();
+
+ Map<String, String> props1 = new HashMap<>();
+ props1.put("key1", "value1");
+ props1.put("KEY2", "VALUE2");
+ props1.put("KeY3", "VaLuE3");
+
+ nonBatchProducer.newMessage()
+ .properties(props1)
+ .value("non-batch-message")
+ .send();
+
+ Message<byte[]> peekedMessage = admin.topics().peekMessages(topicName, "sub-peek", 1).get(0);
+ assertEquals(new String(peekedMessage.getData()), "non-batch-message");
+ assertEquals(peekedMessage.getProperties().size(), 3);
+ assertEquals(peekedMessage.getProperties().get("key1"), "value1");
+ assertEquals(peekedMessage.getProperties().get("KEY2"), "VALUE2");
+ assertEquals(peekedMessage.getProperties().get("KeY3"), "VaLuE3");
+
+ admin.topics().truncate(topicName);
+
+ // Test batch messages
+ @Cleanup
+ Producer<String> batchProducer = pulsarClient.newProducer(Schema.STRING)
+ .topic(topicName)
+ .enableBatching(true)
+ .batchingMaxPublishDelay(1, TimeUnit.SECONDS)
+ .batchingMaxMessages(2)
+ .create();
+
+ Map<String, String> props2 = new HashMap<>();
+ props2.put("batch-key1", "batch-value1");
+ props2.put("BATCH-KEY2", "BATCH-VALUE2");
+ props2.put("BaTcH-kEy3", "BaTcH-vAlUe3");
+
+ batchProducer.newMessage()
+ .properties(props2)
+ .value("batch-message-1")
+ .sendAsync();
+
+ batchProducer.newMessage()
+ .properties(props2)
+ .value("batch-message-2")
+ .send();
+
+ List<Message<byte[]>> peekedMessages = admin.topics().peekMessages(topicName, "sub-peek", 2);
+ assertEquals(peekedMessages.size(), 2);
+
+ for (int i = 0; i < 2; i++) {
+ Message<byte[]> batchMessage = peekedMessages.get(i);
+ assertEquals(new String(batchMessage.getData()), "batch-message-" + (i + 1));
+ assertEquals(batchMessage.getProperties().size(),
+ 3 + 2 // 3 properties from the message + 2 properties from the batch
+ );
+ assertEquals(batchMessage.getProperties().get("X-Pulsar-num-batch-message"), "2");
+ assertNotNull(batchMessage.getProperties().get("X-Pulsar-batch-size"));
+ assertEquals(batchMessage.getProperties().get("batch-key1"), "batch-value1");
+ assertEquals(batchMessage.getProperties().get("BATCH-KEY2"), "BATCH-VALUE2");
+ assertEquals(batchMessage.getProperties().get("BaTcH-kEy3"), "BaTcH-vAlUe3");
+ }
+ }
}