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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
