[GitHub] [pulsar] hrzzzz commented on a diff in pull request #21177: [fix][txn] fix the consumer stuck due to deduplicated messages in pending ack state

["hrzzzz (via GitHub)" <gi...@apache.org>]

hrzzzz commented on code in PR #21177:
URL: https://github.com/apache/pulsar/pull/21177#discussion_r1332553012


##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java:
##########
@@ -254,6 +254,62 @@ private void testFilterMsgsInPendingAckStateWhenConsumerDisconnect(boolean enabl
         Assert.assertEquals(receiveCounter, count / 2);
     }
 
+    @Test
+    private void testMsgsInPendingAckStateWouldNotGetTheConsumerStuck() throws Exception {
+        final String topicName = NAMESPACE1 + "/testMsgsInPendingAckStateWouldNotGetTheConsumerStuck";
+        final String subscription = "test";
+
+        Producer<Integer> producer = pulsarClient.newProducer(Schema.INT32)
+                .topic(topicName)
+                .create();
+        Consumer<Integer> consumer = pulsarClient.newConsumer(Schema.INT32)
+                .topic(topicName)
+                .subscriptionName(subscription)
+                .subscriptionType(SubscriptionType.Shared)
+                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+                .subscribe();
+
+        int numStep1Receive = 2, numStep2Receive = 2, numStep3Receive = 2;
+        int numTotalMessage = numStep1Receive + numStep2Receive + numStep3Receive;
+
+        for (int i = 0; i < numTotalMessage; i++) {
+            producer.send(i);
+        }
+
+        Transaction step1Txn = getTxn();
+        Transaction step2Txn = getTxn();
+
+        // Step 1, try to consume some messages but do not commit the transaction
+        for (int i = 0; i < numStep1Receive; i++) {
+            consumer.acknowledgeAsync(consumer.receive().getMessageId(), step1Txn).get();
+        }
+
+        // Step 2, try to consume some messages and commit the transaction
+        for (int i = 0; i < numStep2Receive; i++) {
+            consumer.acknowledgeAsync(consumer.receive().getMessageId(), step2Txn).get();
+        }
+
+        // commit step2Txn
+        step2Txn.commit().get();
+
+        // close and re-create consumer
+        consumer.close();
+        Consumer<Integer> consumer2 = pulsarClient.newConsumer(Schema.INT32)
+                .topic(topicName)
+                .receiverQueueSize(numStep3Receive)
+                .subscriptionName(subscription)
+                .subscriptionType(SubscriptionType.Shared)
+                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+                .subscribe();
+
+        // Step 3, try to consume the rest messages and should receive all of them
+        for (int i = 0; i < numStep3Receive; i++) {
+            // should get the message instead of timeout
+            Message<Integer> msg = consumer2.receive(10, TimeUnit.SECONDS);

Review Comment:
   fixed



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org