massakam commented on a change in pull request #1977: Enable listener to
receive messages even if receiver queue size is zero
URL: https://github.com/apache/incubator-pulsar/pull/1977#discussion_r196277383
##########
File path:
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
##########
@@ -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);
Review comment:
It is because that a flow command is sent before the listener finishes
processing of the message if existing logic is used. The flow command should be
sent after the listener completes processing of the message.
In addition, I think that it is not preferable to add the message to
`incommingMessages` even though the queue size is zero.
----------------------------------------------------------------
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
