merlimat closed pull request #1977: Enable listener to receive messages even if
receiver queue size is zero
URL: https://github.com/apache/incubator-pulsar/pull/1977
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/.gitignore b/.gitignore
index b009375637..ed5a8b63ea 100644
--- a/.gitignore
+++ b/.gitignore
@@ -7,6 +7,7 @@ logs
/data
pulsar-broker/tmp.*
pulsar-broker/src/test/resources/log4j2.yaml
+pulsar-functions/worker/test-tenant/
*.log
*.versionsBackup
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ZeroQueueSizeTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ZeroQueueSizeTest.java
index 725fa46927..2859047fae 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ZeroQueueSizeTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ZeroQueueSizeTest.java
@@ -20,8 +20,12 @@
import static org.testng.Assert.assertEquals;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
+import com.google.common.collect.Lists;
+
import org.apache.pulsar.broker.service.BrokerTestBase;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.Consumer;
@@ -122,6 +126,50 @@ public void zeroQueueSizeNormalConsumer() throws
PulsarClientException {
}
}
+ @Test()
+ public void zeroQueueSizeConsumerListener() throws Exception {
+ String key = "zeroQueueSizeConsumerListener";
+
+ // 1. Config
+ final String topicName = "persistent://prop/use/ns-abc/topic-" + key;
+ final String subscriptionName = "my-ex-subscription-" + key;
+ final String messagePredicate = "my-message-" + key + "-";
+
+ // 2. Create Producer
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName)
+ .enableBatching(false)
+ .messageRoutingMode(MessageRoutingMode.SinglePartition)
+ .create();
+
+ // 3. Create Consumer
+ List<Message<byte[]>> messages = Lists.newArrayList();
+ CountDownLatch latch = new CountDownLatch(totalMessages);
+ ConsumerImpl<byte[]> consumer = (ConsumerImpl<byte[]>)
pulsarClient.newConsumer().topic(topicName)
+
.subscriptionName(subscriptionName).receiverQueueSize(0).messageListener((cons,
msg) -> {
+ assertEquals(((ConsumerImpl) cons).numMessagesInQueue(),
0);
+ synchronized(messages) {
+ messages.add(msg);
+ }
+ log.info("Consumer received: " + new
String(msg.getData()));
+ latch.countDown();
+ }).subscribe();
+
+ // 3. producer publish messages
+ for (int i = 0; i < totalMessages; i++) {
+ String message = messagePredicate + i;
+ log.info("Producer produced: " + message);
+ producer.send(message.getBytes());
+ }
+
+ // 4. Receiver receives the message
+ latch.await();
+ assertEquals(consumer.numMessagesInQueue(), 0);
+ assertEquals(messages.size(), totalMessages);
+ for (int i = 0; i < messages.size(); i++) {
+ assertEquals(new String(messages.get(i).getData()),
messagePredicate + i);
+ }
+ }
+
@Test()
public void zeroQueueSizeSharedSubscription() throws PulsarClientException
{
String key = "zeroQueueSizeSharedSubscription";
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
index 056b17e9f9..10fe6ff785 100644
---
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
@@ -485,7 +485,8 @@ public void connectionOpened(final ClientCnx cnx) {
AVAILABLE_PERMITS_UPDATER.set(this, 0);
// For zerosize queue : If the connection is reset and
someone is waiting for the messages
// or queue was not empty: send a flow command
- if (waitingOnReceiveForZeroQueueSize ||
(conf.getReceiverQueueSize() == 0 && currentSize > 0)) {
+ if (waitingOnReceiveForZeroQueueSize
+ || (conf.getReceiverQueueSize() == 0 &&
(currentSize > 0 || listener != null))) {
sendFlowPermitsToBroker(cnx, 1);
}
} else {
@@ -678,6 +679,9 @@ void messageReceived(MessageIdData messageId, ByteBuf
headersAndPayload, ClientC
log.debug("[{}][{}] Ignoring message as it was already being
acked earlier by same consumer {}/{}",
topic, subscription, msgId);
}
+ if (conf.getReceiverQueueSize() == 0) {
+ increaseAvailablePermits(cnx);
+ }
return;
}
@@ -722,12 +726,12 @@ void messageReceived(MessageIdData messageId, ByteBuf
headersAndPayload, ClientC
// if the conf.getReceiverQueueSize() is 0 then discard
message if no one is waiting for it.
// if asyncReceive is waiting then notify callback without
adding to incomingMessages queue
unAckedMessageTracker.add((MessageIdImpl)
message.getMessageId());
- boolean asyncReceivedWaiting = !pendingReceives.isEmpty();
- if ((conf.getReceiverQueueSize() != 0 ||
waitingOnReceiveForZeroQueueSize) && !asyncReceivedWaiting) {
- incomingMessages.add(message);
- }
- if (asyncReceivedWaiting) {
+ if (!pendingReceives.isEmpty()) {
notifyPendingReceivedCallback(message, null);
+ } else if (conf.getReceiverQueueSize() != 0 ||
waitingOnReceiveForZeroQueueSize) {
+ incomingMessages.add(message);
+ } else if (conf.getReceiverQueueSize() == 0 && listener !=
null) {
+ triggerZeroQueueSizeListener(message);
}
} finally {
lock.readLock().unlock();
@@ -754,7 +758,7 @@ void messageReceived(MessageIdData messageId, ByteBuf
headersAndPayload, ClientC
msgMetadata.recycle();
}
- if (listener != null) {
+ if (listener != null && conf.getReceiverQueueSize() != 0) {
// Trigger the notification on the message listener in a separate
thread to avoid blocking the networking
// thread while the message processing happens
listenerExecutor.execute(() -> {
@@ -816,6 +820,27 @@ void notifyPendingReceivedCallback(final Message<T>
message, Exception exception
}
}
+ private void triggerZeroQueueSizeListener(final Message<T> message) {
+ checkArgument(conf.getReceiverQueueSize() == 0);
+ checkNotNull(listener, "listener can't be null");
+ checkNotNull(message, "unqueued message can't be null");
+
+ listenerExecutor.execute(() -> {
+ stats.updateNumMsgsReceived(message);
+ try {
+ if (log.isDebugEnabled()) {
+ log.debug("[{}][{}] Calling message listener for unqueued
message {}", topic, subscription,
+ message.getMessageId());
+ }
+ listener.received(ConsumerImpl.this, message);
+ } catch (Throwable t) {
+ log.error("[{}][{}] Message listener error in processing
unqueued message: {}", topic, subscription,
+ message.getMessageId(), t);
+ }
+ increaseAvailablePermits(cnx());
+ });
+ }
+
void receiveIndividualMessagesFromBatch(MessageMetadata msgMetadata,
ByteBuf uncompressedPayload,
MessageIdData messageId, ClientCnx cnx) {
int batchSize = msgMetadata.getNumMessagesInBatch();
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
