[client] Optimize pause when creating sub consumers in multi-topic consumer

[Jason918]

Motivation

Currently, if we subscribe new partitions/topics when multiTopicConsumer is paused, the sub consumers will start in non-pause mode and then changed to pause.

Modifications

Set startPause config when creating sub consumers.

Verifying this change

• Make sure that the change passes the CI checks.

This change is already covered by existing tests, such as

• org.apache.pulsar.client.api.PartitionedProducerConsumerTest#testPauseAndResume
• org.apache.pulsar.client.api.SimpleProducerConsumerTest#testPauseAndResume
• org.apache.pulsar.client.impl.ZeroQueueSizeTest#testPauseAndResume

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

Internal optimization.

[Jason918]

/pulsarbot run-failure-checks

[Jason918]

/pulsarbot run-failure-checks

[Jason918]

/pulsarbot run-failure-checks

[codelipenghui]

LGTM, is it able to add a test?

[aloyszhang]

@Jason918
After creating a new ConsumerImpl for a topic partition, before sending FLOW command, client will check

if (!(firstTimeConnect && hasParentConsumer) && getCurrentReceiverQueueSize() != 0) {
                    increaseAvailablePermits(cnx, getCurrentReceiverQueueSize());
                }

hasParentConsumer is always true for multiTopicConsumer, so the new created ConsumerImpl will not send FLOW command to broker.

BTW, #11974 introduce startPaused for creating new ConsumerImpl, so this pull request is a nice optimize for me.

Left a comment, PTAL.

[Jason918]

After creating a new ConsumerImpl for a topic partition, before sending FLOW command, client will check

if (!(firstTimeConnect && hasParentConsumer) && getCurrentReceiverQueueSize() != 0) {
                    increaseAvailablePermits(cnx, getCurrentReceiverQueueSize());
                }

hasParentConsumer is always true for multiTopicConsumer, so the new created ConsumerImpl will not send FLOW command to broker.

Great point. I have updated the motivation description.