Fix ack failure on message listener in multi topics consumer

[nkurihar]

Fixes #446

Motivation

We found an ack failure issue: #446

MessageListeners can start to process messages before subscribing all child topics are completed in "multi topics" (e.g. partitioned, list, regex) cases.
If acknowledge() is called in messageListeners (it is very typical usage) before subscribing completion, it will fail with AlreadyClosed error since the state of the parent consumer is not Ready yet:

pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc

Lines 643 to 648 in 8b2753a

	void MultiTopicsConsumerImpl::acknowledgeAsync(const MessageId& msgId, ResultCallback callback) {
	if (state_ != Ready) {
	interceptors_->onAcknowledge(Consumer(shared_from_this()), ResultAlreadyClosed, msgId);
	callback(ResultAlreadyClosed);
	return;
	}

That results in ack holes, and finally full backlog.

Modifications

• MultiTopicsConsumerImpl: Pause messageListeners at first by setting startPaused to true and resume them after subscribing all child topics are completed.
• ConsumerImpl: Delete unused code.
  • It tries to skip sendFlowPermitsToBroker() at the first connection.
    • The motivation seems same as this PR, i.e. to prevent messageListeners from starting before subscribing completion.
  • However, actually it does not work so far, the variable firstTime looks always false at

    pulsar-client-cpp/lib/ConsumerImpl.cc

    Line 316 in 8b2753a

    if (consumerTopicType_ == NonPartitioned || !firstTime) {

    because it becomes false at

    pulsar-client-cpp/lib/ConsumerImpl.cc

    Lines 297 to 299 in 8b2753a

    	if (firstTime) {
    	firstTime = false;
    	}

    • Also it is static variable in a method that is shared by all ConsumerImpl instances, whereas it has no chance of returning to true once it becomes false.
  • It seems better that child consumers don't have to care about their parents and are completely independent of them.

Verifying this change

• Make sure that the change passes the CI checks.

Specific tests for this issue is difficult to implement because it does not occur every time.

Documentation

• doc-required
  (Your PR needs to update docs and you will update later)

• doc-not-needed
  (Please explain why)

• doc
  (Your PR contains doc changes)

• doc-complete
  (Docs have been already added)