Allows consumer retrieve the sequence id that the producer set. (apac…

…he#4645) * Allows consumer retrieve the sequence id that the producer set. * fix comments.
lionelcao · Jul 2, 2019 · d107f67 · d107f67
1 parent fc9c376
commit d107f67
Show file tree

Hide file tree

Showing 5 changed files with 131 additions and 0 deletions.
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageTest.java
@@ -737,5 +737,69 @@ public void testOrderingOfKeyBasedBatchMessageContainer() throws PulsarClientExc
         producer.close();
     }
 
+    @Test(dataProvider = "containerBuilder")
+    private void testRetrieveSequenceIdGenerated(BatcherBuilder builder) throws Exception {
+
+        int numMsgs = 10;
+        final String topicName = "persistent://prop/ns-abc/testRetrieveSequenceIdGenerated-" + UUID.randomUUID();
+        final String subscriptionName = "sub-1";
+
+        Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicName).subscriptionName(subscriptionName)
+                .subscriptionType(SubscriptionType.Shared).subscribe();
+
+        Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName)
+                .batchingMaxPublishDelay(5, TimeUnit.SECONDS).batchingMaxMessages(numMsgs).enableBatching(true)
+                .batcherBuilder(builder)
+                .create();
+
+        List<CompletableFuture<MessageId>> sendFutureList = Lists.newArrayList();
+        for (int i = 0; i < numMsgs; i++) {
+            byte[] message = ("my-message-" + i).getBytes();
+            sendFutureList.add(producer.sendAsync(message));
+        }
+        FutureUtil.waitForAll(sendFutureList).get();
+
+        for (int i = 0; i < numMsgs; i++) {
+            Message<byte[]> received = consumer.receive();
+            Assert.assertEquals(received.getSequenceId(), i);
+            consumer.acknowledge(received);
+        }
+
+        producer.close();
+        consumer.close();
+    }
+
+    @Test(dataProvider = "containerBuilder")
+    private void testRetrieveSequenceIdSpecify(BatcherBuilder builder) throws Exception {
+
+        int numMsgs = 10;
+        final String topicName = "persistent://prop/ns-abc/testRetrieveSequenceIdSpecify-" + UUID.randomUUID();
+        final String subscriptionName = "sub-1";
+
+        Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicName).subscriptionName(subscriptionName)
+                .subscriptionType(SubscriptionType.Shared).subscribe();
+
+        Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName)
+                .batchingMaxPublishDelay(5, TimeUnit.SECONDS).batchingMaxMessages(numMsgs).enableBatching(true)
+                .batcherBuilder(builder)
+                .create();
+
+        List<CompletableFuture<MessageId>> sendFutureList = Lists.newArrayList();
+        for (int i = 0; i < numMsgs; i++) {
+            byte[] message = ("my-message-" + i).getBytes();
+            sendFutureList.add(producer.newMessage().sequenceId(i + 100).value(message).sendAsync());
+        }
+        FutureUtil.waitForAll(sendFutureList).get();
+
+        for (int i = 0; i < numMsgs; i++) {
+            Message<byte[]> received = consumer.receive();
+            Assert.assertEquals(received.getSequenceId(), i + 100);
+            consumer.acknowledge(received);
+        }
+
+        producer.close();
+        consumer.close();
+    }
+
     private static final Logger LOG = LoggerFactory.getLogger(BatchMessageTest.class);
 }
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java
@@ -148,6 +148,10 @@ static <T> MessageImpl<T> create(MessageMetadata.Builder msgMetadataBuilder, Byt
             msgMetadataBuilder.setEventTime(singleMessageMetadata.getEventTime());
         }
 
+        if (singleMessageMetadata.hasSequenceId()) {
+            msgMetadataBuilder.setSequenceId(singleMessageMetadata.getSequenceId());
+        }
+
         this.schema = schema;
     }
 

diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/api/proto/PulsarApi.java b/pulsar-common/src/main/java/org/apache/pulsar/common/api/proto/PulsarApi.java
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java
@@ -1170,6 +1170,10 @@ public static ByteBuf serializeSingleMessageInBatchWithPayload(PulsarApi.Message
             singleMessageMetadataBuilder.setEventTime(msgBuilder.getEventTime());
         }
 
+        if (msgBuilder.hasSequenceId()) {
+            singleMessageMetadataBuilder.setSequenceId(msgBuilder.getSequenceId());
+        }
+
         try {
             return serializeSingleMessageInBatchWithPayload(singleMessageMetadataBuilder, payload, batchBuffer);
         } finally {

diff --git a/pulsar-common/src/main/proto/PulsarApi.proto b/pulsar-common/src/main/proto/PulsarApi.proto
@@ -137,6 +137,8 @@ message SingleMessageMetadata {
 	optional bool partition_key_b64_encoded = 6 [ default = false ];
     // Specific a key to overwrite the message key which used for ordering dispatch in Key_Shared mode.
 	optional bytes ordering_key = 7;
+	// Allows consumer retrieve the sequence id that the producer set.
+	optional uint64 sequence_id = 8;
 }
 
 enum ServerError {