[GitHub] [pulsar] Jason918 commented on a diff in pull request #15852: [fix][client] `ZeroQueueConsumer` may not increase the permit after unloading the topic.

[GitBox <gi...@apache.org>]

Jason918 commented on code in PR #15852:
URL: https://github.com/apache/pulsar/pull/15852#discussion_r886266398


##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ZeroQueueConsumerImpl.java:
##########
@@ -85,10 +85,13 @@ protected Message<T> internalReceive() throws PulsarClientException {
     @Override
     protected CompletableFuture<Message<T>> internalReceiveAsync() {
         CompletableFuture<Message<T>> future = super.internalReceiveAsync();
+        waitingOnReceiveForZeroQueueSize = true;
         if (!future.isDone()) {
-            // We expect the message to be not in the queue yet
             increaseAvailablePermits(cnx());
         }
+        future.whenComplete((message, ex) -> {
+            waitingOnReceiveForZeroQueueSize = false;

Review Comment:
   Consider the case that `internalReceiveAsync` is called concurrently, the state of `waitingOnReceiveForZeroQueueSize` will not be right for the second time.
   
   And FYI `waitingOnReceiveForZeroQueueSize` is actually protected by `zeroQueueLock` in `internalReceive()`.



##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ZeroQueueSizeTest.java:
##########
@@ -546,4 +548,48 @@ public void testPauseAndResumeNoReconnection() throws Exception {
             assertEquals(receivedMessages.get(i).intValue(), i);
         }
     }
+
+    @Test(timeOut = 30000)
+    public void testReceiveAsyncWhenUnloadTopic() throws PulsarClientException, ExecutionException, InterruptedException, PulsarAdminException {
+        final String topicName = "persistent://prop/cluster/namespace/topic-" + UUID.randomUUID();
+        int messageNumber = 500;
+        List<MessageId> messages = Collections.synchronizedList(new ArrayList<>());
+        @Cleanup
+        Consumer<byte[]> subscribe = pulsarClient.newConsumer()
+                .topic(topicName)
+                .subscriptionName("sub-11")
+                .subscriptionType(SubscriptionType.Shared)
+                .receiverQueueSize(0)
+                .consumerName("con-11")
+                .subscribe();
+
+        @Cleanup
+        Producer<byte[]> producer = pulsarClient.newProducer()
+                .enableBatching(false)
+                .messageRoutingMode(MessageRoutingMode.RoundRobinPartition)
+                .topic(topicName).create();
+        for (int i = 0; i < messageNumber; i++) {
+            MessageId msgId = producer.newMessage()
+                    .value(UUID.randomUUID().toString().getBytes(StandardCharsets.UTF_8))
+                    .send();
+            messages.add(msgId);
+        }
+
+        boolean isUnload = false;
+        do {
+            Message<byte[]> message;
+            try {
+                message = subscribe.receiveAsync().get(5, TimeUnit.SECONDS);
+                if (message != null) {
+                    subscribe.acknowledge(message);
+                    messages.remove(message.getMessageId());
+                }
+            } catch (TimeoutException e) {
+            }
+            if (!isUnload) {
+                admin.topics().unload(topicName);
+                isUnload = true;
+            }
+        } while (messages.size() != 0);

Review Comment:
   Try a concurrent case for `subscribe.receiveAsync()` ?



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