fix consumer stuck when batchReceivePolicy maxNumMessages > maxReceiverQueueSize

[hangc0276]

Fix #6854

Bug description

The consumer stuck due to hasEnoughMessagesForBatchReceive checking:

protected boolean hasEnoughMessagesForBatchReceive() {
        if (batchReceivePolicy.getMaxNumMessages() <= 0 && batchReceivePolicy.getMaxNumBytes() <= 0) {
            return false;
        }
        return (batchReceivePolicy.getMaxNumMessages() > 0 && incomingMessages.size() >= batchReceivePolicy.getMaxNumMessages())
                || (batchReceivePolicy.getMaxNumBytes() > 0 && INCOMING_MESSAGES_SIZE_UPDATER.get(this) >= batchReceivePolicy.getMaxNumBytes());
    }

Changes

When batchReceivePolicy maxNumMessages > maxReceiverQueueSize, we force batch receivePolicy maxNumMessages to maxReceiverQueueSize to avoid consumer stuck when checking hasEnoughMessagesForBatchReceive

[codelipenghui]

I think we should introduce an exception when the maxNumMessages greater than queue size when creating consumers. This will be more transparent to users.

[codelipenghui]

We also need to add a check for DEFAULT_POLICY since users may specify the queue size that lower that maxNumMessages of the DEFAULT_POLICY.

[avimas]

The above solution is correct but users can get confused seeing their batchRecieve policy configuration has changed
Why not changing the function hasEnoughMessagesForBatchReceive to return true when
incomingMessages.size() >= maxReceiverQueueSize ?

protected boolean hasEnoughMessagesForBatchReceive() {
if (batchReceivePolicy.getMaxNumMessages() <= 0 && batchReceivePolicy.getMaxNumBytes() <= 0) {
return false;
}
return (batchReceivePolicy.getMaxNumMessages() > 0 && incomingMessages.size() >= batchReceivePolicy.getMaxNumMessages())
|| (batchReceivePolicy.getMaxNumBytes() > 0 && INCOMING_MESSAGES_SIZE_UPDATER.get(this) >= batchReceivePolicy.getMaxNumBytes()) || (incomingMessages.size() >= maxReceiverQueueSize);
}

What do you think ?

[codelipenghui]

This will cause batch message return to the user earlier right? This will introduce another problem, users will get the batch message before reaching maxNumMessage. It's better to tell users the receiver queue size needs to greater than the maxNumMessage of the batch receive policy.

[avimas]

I completely agree that throwing an exception is a better choice

[psilos]

can we also add unit tests with the desired behaviour ?

[hangc0276]

Thanks for your feedback, I have add the unit tests for all kinds of exception case, please take a look.

[codelipenghui]

@hangc0276 Jia has started a discussion in the email thread for cutting 2.5.2 and it's better to include this fix in 2.5.2. If you have time, please help take a look at the comments, so that we can merge this PR.

[hangc0276]

@hangc0276 Jia has started a discussion in the email thread for cutting 2.5.2 and it's better to include this fix in 2.5.2. If you have time, please help take a look at the comments, so that we can merge this PR.

sorry，I will fix this pr soon. When i add test case for the batchReceivePolicy, I found another bugs, and the consumer still stuck when reset the maxNumMessages to receiverQueueSize. I will fix it soon.

[hangc0276]

I think we should introduce an exception when the maxNumMessages greater than queue size when creating consumers. This will be more transparent to users.

I have throw a log when reset the batch receive policy, please take a look, thanks.

[hangc0276]

@codelipenghui When batch receive triggered by timeout, it should trigger send permit to broker server in MultiTopicsConsumerImpl. Without the trigger, it will also stuck the consumer.

[hangc0276]

/pulsarbot run-failure-checks