[PIP-74] Dependency of consumer client memory limit: support dynamic limit of consumer receiver queue

[Jason918]

Motivation

This is part of the work for PIP 74
We need dynamic update currentReceiverQueue to control client memory.

Modifications

Add getter and setter method for ConsumerBase#maxReceiverQueueSize.

• For ConsumerImpl, we need update availablePermits together.
• For MultiTopicsConsumerImpl, we need update inner consumers together and trigger paused consumers.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• org.apache.pulsar.client.impl.DynamicReceiverQueueSizeTest
• ConsumerImplTest#testMaxReceiverQueueSize

Does this pull request potentially affect one of the following parts:

If yes was chosen, please highlight the changes

• Dependencies (does it add or upgrade a dependency): (no)
• The public API: (no)
• The schema: (no)
• The default values of configurations: (no)
• The wire protocol: (no)
• The rest endpoints: (no)
• The admin cli options: (no)
• Anything that affects deployment: (no)

Documentation

Check the box below and label this PR (if you have committer privilege).

Need to update docs?

• no-need-doc

[Jason918]

/pulsarbot run-failure-checks

[michaeljmarshall]

Looks good to me. Left one question.

[mattisonchao]

I think this PR changed the behaviour of the original Consumer#receiverQueueSize,
it looks seem will auto-scale the queue size, maybe we can change the receiverQueueSize to initialReceiverQueueSize and add a doc to explain it.

I'm not sure about this, Please let me know what you think. :)

[michaeljmarshall]

I think we need to handle the ZeroQueueConsumerImpl better since that case is not eligible for dynamic queue resizing, at least based on the current class design.

@mattisonchao - good point about the names. However, I think I have a different perspective on what the final names should be.

We currently set the initial queue size with a method named receiverQueueSize on the consumer builder and then we update it with a method named setMaxReceiverQueueSize on the consumer. I assume that this difference comes from the fact that ConsumerBase has a protected variable named maxReceiverQueueSize. I think that our API should be consistent when possible, and since receiverQueueSize has been on the builder since 2018, I'd prefer to update the new setter method to setReceiverQueueSize. That might mean we need to update the protected variable named maxReceiverQueueSize. What is your perspective @Jason918?

(I am requesting changes because of the ZeroQueueConsumerImpl issue and not because of the naming discussion.)

[Jason918]

it looks seem will auto-scale the queue size

Yes, it's exactly what PIP-74 want to do.

maybe we can change the receiverQueueSize to initialReceiverQueueSize and add a doc to explain it.

It's actually would be the max value of auto-scaled receiver queue size.

[Jason918]

I think we need to handle the ZeroQueueConsumerImpl better since that case is not eligible for dynamic queue resizing, at least based on the current class design.

Good catch.

I think that our API should be consistent when possible, and since receiverQueueSize has been on the builder since 2018, I'd prefer to update the new setter method to setReceiverQueueSize

Good point. Previously, I use maxReceiverQueueSize because receiverQueueSize can be regarded as the current size of incomingMessages, actually I think receiverQueueCapacity maybe a proper name than receiverQueueSize. So let's leave it be and not change the concept at this moment.

[michaeljmarshall]

LGTM. Thanks for the explanations @Jason918, I didn't understand the feature at first--I thought the dynamic part was going to be exposed to the end user.