[PIP-74] Support auto scaled consumer receiver queue

[Jason918]

Motivation

1. Pick a proper receiver queue size is not an easy thing. If the value is too small, it impact the throughput, and if the value is too large, it consumes too many memory. With default set up, the queue size is 1000, and the max message size is 5MB, this means maximum of 5GB memory occupation.

2. This is part of the work for PIP 74. We need auto scale currentReceiverQueue to control client memory.

Modifications

Add optional autoScaledReceiverQueueSizeEnabled for consumer client.

Previous receiverQueueSize is the max value of this auto scaled queue size.

Every time the client will try to double the size of currentReceiverQueue if it limits message throughput. Currently, it's determined by the following two conditions in exact order:
A) Current receiver queue (ConsumerBase#incomingMessages) is full after we put a message into it. (it's marked by scaleReceiverQueueHint as true).
B) Application wants process more messages but the receiver queue is empty. (expectMoreIncomingMessages is called in this PR).

The queue size won't grow if we got new messages during A and B. So if assume current receiver queue size is 10, and the timeline would be like

1. Consumer send flow command to client, and client receive 10 messages, so scaleReceiverQueueHint is marked as true.
2. Application calls receive() repeatedly and processed all 10 messages. And in the meanwhile no new message is sent to client[1].
3. Receiver queue size will be doubled to 20.
4. ...

NOTE:
Condition A here is slightly different from original design in PIP-74 described as "there are messages pending to be sent for this consumer" (let's refer it as Condition X). This is proposed with these consideration:
a) We can assume that when receiver queue is full, the chance of broker have no more messages is insignificant in practice.
b) If we accept a), then Condition A implies Condition X, as new messages are pending because local queue size is full. But it's NOT vice versa.
c) Once fact we should accept is that we don't need expand queue size every time it limits the throughput as some slight and occasional delay is acceptable and won't affect overall throughput. And by replacing Condition A with Condition X, we can reduce the sensitivity of the expansion.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

(example:)

• org.apache.pulsar.client.impl.AutoScaledReceiverQueueSizeTest

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: (yes, add a config for consumer builder)
• 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

[Jason918]

Here is a test with bin/pulsar-perf on local standalone server to confirm the effect of this PR.

Non-partitioned topic consumer

bin/pulsar-perf consume -aq -q 1000000 persistent://public/default/test
bin/pulsar-perf produce -r $RATE -s 128 -bm 0 -time 60 persistent://public/default/test

RATE	Final receiver queue size
1	2
10	2
100	4
1000	32
10000	512

3-partitioned topic consumer

bin/pulsar-perf consume -aq -q 1000000 persistent://public/default/multi-partitions
bin/pulsar-perf produce -r $RATE -s 128 -bm 0 -time 60 -np 3 persistent://public/default/multi-partitions

RATE	Final receiver queue size (MultiTopicConsumer)	Sub-consumers receiver queue size
1	3	1,1,1
10	3	1,1,1
100	6	2,1,1
1000	12	16,32,16
5000	48	128,128,128
10000	96	1024,512,512
20000	192	1024,512,512
40000	384	1024,1024,1024

[Jason918]

/pulsarbot run-failure-checks

[Jason918]

/pulsarbot run-failure-checks

[Jason918]

/pulsarbot run-failure-checks

[Jason918]

/pulsarbot run-failure-checks

[Jason918]

/pulsarbot run-failure-checks

[Jason918]

@merlimat @hangc0276 @315157973 @codelipenghui PTAL

[Jason918]

/pulsarbot run-failure-checks

[HQebupt]

👍

[StevenLuMT]

Good job

[Jason918]

@codelipenghui @315157973 @hangc0276 @aloyszhang Ping~

[gaozhangmin]

LGTM

[Jason918]

/pulsarbot run-failure-checks

[lordcheng10]

LGTM