Fixes #7851 

### Motivation

`clear()` methods of `BatchAcknowledgementTracker` and `UnAckedMessageTrackerEnabled` are not thread-safe.

### Modifications

Acquire a mutex in these `clear()` methods.

* fix: add message id to send message in cpp

* test: fix produce consume test to check whether message id was changed or not

Fixes #7168 

### Motivation

When a pattern consumer is blocked by `receive()`, the `mutex_` will be held until new messages arrived. If the auto discovery timer task found new topics and tried to subscribe them, `mutex_` must be acquired first, then the deadlock happened.

### Modifications

- Release the `mutex_` after the consumer's state was verified.
- Change unit tests to verify that new topics could be subscribed when the consumer's blocked by `receive(Message&)` or `receive(Message&, int)` methods.

### Verifying this change

- [ ] Make sure that the change passes the CI checks.

This change is already covered by existing tests, such as *BasicEndToEndTest.testPatternMultiTopicsConsumerAutoDiscovery*.

Fixes #6000 

### Motivation

In our production environment, we created a sinker: It combines messages read from Pulsar, and writes the combined data into ClickHouse, which is not good at processing highly concurrent inserting (max. 100 QPS).

In the sinker, we used failover subscription with cumulative ACK. The reason for not using shared subscription is that, individual ACK in C++ client siginificantly reduces throughput. Our analysis shows that huge amount of individual ACK requests results in highly concurrent accessing, which finally results in terrible throughput. BTW, the machines, where we deploy brokers and bookies, are not equipped with SSD. There could be some configuration for brokers or bookies helping improving the situation, but we don't find them.

Therefore, the idea is that combining individual ACK requests within one request. After reading the protocol (PulsarApi.proto:CommandAck), we find that this feature is supported. According to this issue #6000 , Pulsar C++ client SDK does not implement ACK grouping feature, but it's already implemented by Java client.

After discussing with @jiazhai and @sijie , we are authorized to implement this feature and contribute this feature back to community later. It should follow the Java client's behavior and provide similar interfaces that Java client provides.

### Modifications

#### Interfaces

Similar to Java client, C++ client implements three trackers to *cache* ACK requests within a configured time window:

1. `AckGroupingTracker`: this is the base class of the other two trackers, it defines interfaces and provides empty implementation which does not send ACK requests to broker. This tracker is used for non-persistent topic, which actually does not require ACK.
2. `AckGroupingTrackerDisabled`: child class of `AckGroupingTracker`. It does not provide ACK grouping ability, and acts just like the previous individual ACK.
3. `AckGroupingTrackerEnabled`: child class of `AckGroupingTracker`. This is the real implementation of ACK grouping.

The trackers provides following public interfaces:

1. `isDuplicate`: checking if the given message ID exists in the TO-BE-ACKED group.
2. `addAcknowledge` and `addAcknowledgeCumulative`: unlike Java client, which combines these two interfaces into one (`addAcknowledge`), C++ clients provides them for individual and cumulative ACK requests. Such design can provide slightly better performance than if-else implementation.
3. `close`: closing the tracker.
4. `flush` and `flushAndClean`: flushing all pending ACK requests, the later one also resets internal status.

#### Consumer's Configuration

Two new configuration options are added:

1. `ackGroupingTimeMs`: time window in milliseconds for grouping ACK requests. If setting to positive value, ACK requests are grouped and sent to broker every `ackGroupingTimeMs` milliseconds (`AckGroupingTrackerEnabled`). For non-positive values, ACK requests are sent one by one to brokers as before (`AckGroupingTrackerDisabled`). Default is 100.
2. `ackGroupingMaxSize`: maximum number of grouped message IDs. Java client hard-coded it to 1000 for now. However, 1000 is too small in our scenario. Once the grouped message number reaches this limit, the ACK requests will be packed into one and sent to broker, even the ACK grouping deadline (`ackGroupingTimeMs`) is not reached. Non-positive values remove such limit.

**In addition**, these configurations are added into C API as well.

#### Commands for this feature.

A few new command factory interfaces are implemented, just like in Java, incl.

* `newMultiMessageAck`: command object for multi-message ACK requests.
* `peerSupports*`: interfaces for checking versions.

#### Topic Domain

Used to help defining and distinguish non-persistent and persistent topics.

### Verifying this change

- [x] Make sure that the change passes the CI checks.

This PR added 7 unit tests in `BasicEndToEndTest`, all of them start with `testAckGroupingTracker`. They cover the ACK grouping tracker's logic.

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

NO

  - 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

  - Does this pull request introduce a new feature? (no)
  - If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
  - If a feature is not applicable for documentation, explain why?
  - If a feature is not documented yet in this PR, please create a followup issue for adding the documentation

Negative acknowledgement runs in the background on a consumer and
triggers redelivery of messages. The tests verify a that messages do
indeed get redelivered, and which messages they are, for the base
case, batching and partitioned consumer.

There's a fundamental dependency on timing in the base case. If 100ms
pass between consumer creation and receiving the last message in first
receive loop, redelivery will be triggered and the order of messages,
as asserted by the test will fail.

This first case can be fixed by moving the negative ack to run after
all messages have been received. However, this can also then fail for
the batch case.

If the negative ack tracker kicks off during the loop to negatively
ack the messages, then the redelivery will happen twice (and possibly
more times depending on how many time it manages to run).

For this reason, if we want the test to be deterministic, we need to
disable the tracker from kicking off redelivery while we send mark the
messages as negatively acked.
Co-authored-by: NIvan Kelly <ikelly@splunk.com>

The topics used in a bunch of the cases in BasicEndToEndTest had no
unique identifier, so if you ran it multiple times against the same
standalone cluster, they would fail, as the topics would already have
some state.

This change adds the unique identifer to the topics.
Co-authored-by: NIvan Kelly <ikelly@splunk.com>

### Motivation

A bunch of tests were using 100ms as the timeout for receive. 100ms is
way too short. The broker for these tests is standalone, so everything
is running in one process on one machine. I/O, locks, GC etc can
easily make reads take longer than 100ms.

We are testing functionality, not performance with these tests, so
I've increased the timeout to 3s.

### Motivation

If messages were sent in batch, every single message would reserve one spot of producer's pending message queue, but only one batched message would be pushed to the queue. Therefore there may exist many unused spots when `ProducerQueueIsFull` happened.

Besides, if a message was too big or failed to be encrypted, `sendAsync` failed immediately but the reserved spot won't be released.

### Modifications

- Add a `bool` return value to `BatchMessageContainer::sendMessages` to indicate whether the batched message was pushed to producer's queue.
- Add a `bool` return value to `BatchMessageContainer::add` to indicate whether the reserved spot should be released. The spot would be retained only if the first message was batched and not sent immediately. The spot would be released when the batched message was pushed to the queue.
- Test sending a batch with a 2-spots pending message queue, one spot for storing the batched message, another spot for preventing `ProducerQueueIsFull` error.
- Test after all batched messages being sent, whether the reserved spots of producer's queue were  0.

### Verifying this change

- [ ] Make sure that the change passes the CI checks.

This change is already covered by existing tests, such as `BatchMessageTest.testSendCallback`, `BatchMessageTest.testPartitionedTopics` and `BasicEndToEndTest.testMessageTooBig`.

### Motivation

The subscription `InitialPosition` is not currently set when using the multi-topic or regex consumers in C++/Python. 

That makes that if you try to start from `MessageId::earliest`, it would be ignored.

* [C++] Subscription InitialPosition is not correctly set on regex consumers

* fix test fail for topic name

* fix `make format`
Co-authored-by: NJia Zhai <zhaijia@apache.org>

### Motivation

If messages were sent to a partitioned topic, the message id's `partition` field was always -1 because SendReceipt command only contains ledger id and entry id.

### Modifications

- Add a `partition` field to `ProducerImpl` and set the `MessageId`'s `partition` field with it in `ackReceived` method later.
- Add a test to check message id in send callback if messages were sent to a partitioned topic.

* Fix bug: sendCallback's 2nd argument was always the default MessageId

* Set batch index for each message's callback of batch

* Add test for message id in send callback

* Ensure all send callbacks completed before ASSERT_EQ

As stated in #6171, change #6113 broke the function instance runner. This change attempts to revert #6113 first to make sure CI back to normal.

Additionally, it fixes and improves a bunch of integration tests. (unfortunately, we have to couple these changes to make sure it pass CI).

Fixes #5676

### Motivation


Expose the redelivery count in the C++ client.

### Modifications

Exposed the redelivery count from the broker in `Message` and `MessageImpl`. Set the counter when receiving messages. Added test.

fix #4994
For non partition topic,

Topic name contains partition suffix "-partition-" and the remaining part follow the partition
suffix is numeric value larger than the number of partition if there's already a partition topic with same
name(the part before suffix "-partition-").
2)Topic name contains partition suffix "-partition-" and the remaining part follow the partition
suffix is numeric value but there isn't a partitioned topic with same name.
For partition topic,
Validation will fail and throw RestException if

There's already a partitioned topic with same topic name and have some of its partition created.
There's already non partition topic with same name and contains partition suffix "-partition-"
followed by numeric value. In this case internal created partition of partitioned topic could override
the existing non partition topic.
This is for non partition topic created before we enforce the check as we will prevent creation of non partition topic with such name which could lost of confusion.
For update partition topic,
Validation will fail if there's already non partition topic with same name and contains partition suffix "-partition-" followed by numeric value X then the new number of partition of that partitioned topic can not be greater than that X else that non partition topic will essentially be overwritten and cause unexpected consequence.

Also removed TopicsConsumerImplTest#testTopicAutoUpdatePartitions in flavor of PartitionedProducerConsumerTest#testAutoUpdatePartitionsForProducerConsumer, as they're both testing auto partition update to consumer/producer. But TopicsConsumerImplTest#testTopicAutoUpdatePartitions was trying to manually create non partitioned topic with name like "xyz-topic-partition-5" by creating new producer of that topic, while 5 is larger than a existing partitioned topic xyz-topic's actual number of partition. Which make the validation fails and shouldn't be encouraged.

* Allow to specify delivery delay in C++ client

* Fixed symbols visibility to make it available for unit tests

* Fixed format

* [Cpp] Fixed negative ack tracker constructor sequence

* Fixed formatting

* Fixed default for neg ack delay on partitioned topics

* Fixed format

### Motivation

The `Send(context.Context, ProducerMessage) error` should return `MessageID`.

```
 SendAsync(context.Context, ProducerMessage, func(ProducerMessage, error))
```

* Not allow use acknowledgeCumulative on Key_shared subscription.

* fix comments

* C++ Client add acknowledgeCumulative not allowed check.

### Motivation

IN PR #4883, we support 1 partitioned topic producer/consumer in java client. this is for Cpp client support.

### Modifications
- change cpp client 
- add unit test

### Verifying this change
New added Ut Passed

 MacOS clang/llvm compiler imcompatible code in Unit test case BasicEndToEndTest.cc #4871

Fixes #4589

Motivation
When using Partition-topic, the logic of redeliver messages will not be triggered when the time of ackTimeout arrives.

This is because the unAckedMessageTrackerPtr_->add(msg.getMessageId()) is not call in the listener handling of partitioned topic in cpp code

* support max message size for cpp and go client
Signed-off-by: Nxiaolong.ran <ranxiaolong716@gmail.com>

* format cpp code
Signed-off-by: Nxiaolong.ran <ranxiaolong716@gmail.com>

* fix maxMessageSize logic
Signed-off-by: Nxiaolong.ran <ranxiaolong716@gmail.com>

* fix comments
Signed-off-by: Nxiaolong.ran <ranxiaolong716@gmail.com>

* fix comments
Signed-off-by: Nxiaolong.ran <ranxiaolong716@gmail.com>

* fix comments
Signed-off-by: Nxiaolong.ran <ranxiaolong716@gmail.com>

* fix comments
Signed-off-by: Nxiaolong.ran <ranxiaolong716@gmail.com>

* fix ci error
Signed-off-by: Nxiaolong.ran <ranxiaolong716@gmail.com>

* fix comments
Signed-off-by: Nxiaolong.ran <ranxiaolong716@gmail.com>

* fix comments
Signed-off-by: Nxiaolong.ran <ranxiaolong716@gmail.com>

* fix comments
Signed-off-by: Nxiaolong.ran <ranxiaolong716@gmail.com>

* Fix C++ consumer unsubscribe when regex doesn't match

* Fixed formatting

* Revert "[cpp client] implement reference count for close() (#3863)"

This reverts commit ee98e8b2.

* Revert Prevent dup consumers and refCount for cpp client

Revert "[cpp client] Bugfix prevent dup consumer for same topic subscription"
Revert "[Issue #3226][cpp client] Prevent dup consumers on same client cnx"

This reverts commit fff02e2a.
This reverts commit 762e0ab9.

* Revert "Feature - implement reference count for ConsumerImpl (#3795)"

This reverts commit ff4db8db.

* Revert Prevent dup consumers and refCount for java client

Revert "Prevent dup consumers on same client cnx with shared subscription"
Revert "[java client] Bugfix prevent dup consumers for same topic subscribe"

This reverts commit 231db030.
This reverts commit fb5dcd9a.

[Issue 2461][pulsar-client-cpp] Modified CMake files and source to enable compilation on Windows (#4071)

Fixes #2461

### Motivation


My motivation was to be able to use the pulsar cpp client on Windows systems. 

### Modifications

There are a number of modifications I needed to make to enable Windows compilation
* Extensive rework of multiple CMakeLists.txt files
* The creation of a PULSAR_PUBLIC define to define symbol visibility in lieue of #pragma GCC visibility push(default). This is because Windows requires specifying __declspec(dllexport) and __declspec(import) explicitly and does not have a #pragma GCC visibility push(default) analogue.
* all calls to usleep or sleep have been replaced with calls to boost::this_thread::sleep()



### Verifying this change

This change is a trivial rework / code cleanup without any test coverage. However, it does introduce a new platform to test on. It is likely that CI checks will have to be created. 

Add reference count feature to keep track of reused instances of a consumer
instance, for more details please see commit ff4db8db.

*Modifications*

  - Add refCount instance variable on ConsumerImpl.
  - Use new safeDecrRefCount() on consumer close() in order to know whether
    effective close call should occur or not.
  - Increment reference count when a previous built consumer instance is being
    used by caller.

*Future considerations*

Thereafter when feature preventing duplicated consumer is made for
PartitionedConsumer, MultiTopicsConsumer and PatternMultiTopicsConsumer,
incrRefCount() member could be turned into a pure virtual method.

* fix bug involving ack timeout in  pulsar cpp client when subscribing to regex topic

* remove newline

* fix indent

* add test

* addressing comments

* fix formatting

* fix formatting

Same fix as #3746 but for cpp client.

  - Filter consumers for the same topic name.
  - Add test to verify that same subscription names with different topics are
    allowed to be different consumers subscription instead of reused.

* Negative acks in C++ client

* Fixed formatting

* Fixed api docs

* In C++ allow messages that compress to <5mb to be sent with batching enabled

* Fixed related test

Use testDupConsumersOnSharedModeNotThrowsExcOnUnsubscribe to avoid interference with other tests running in parallel (#3521)

* Use testDupConsumersOnSharedModeNotThrowsExcOnUnsubscribe to avoid interference with other tests running in parallel

* Fixed formatting

* Replaced boost::bind with std::bind

* Fixed merging with master

* std::bind doesn't work with overloaded functions

* Go back to asio thread to executor service

* Use proper ref-count increase for WaitForCallback utility

* Replace boost::mutex and boost::thread with std::mutex and std::thread

* Added missing mutex header

* Added missing vector header

* Removed extra tab

* Added missing <atomic> include

* Missing <memory> includes

Prevent same consumer subscription over the same client connection for cpp
client. For more details please see explanation on commit 44e1a23.

### Motivation

Fix #3226 for cpp client.

### Modifications

Add check for `ClientImpl.cc` to know whether new shared consumer subscribe is already present
on consumers vector or not.

For further explanation please see #3312 

### Verifying this change

This change added tests and can be verified as follows:

  - Add unit test asserting if duplicated consumers are allowed for shared
    subscription.
  - Add test preventing broker metadata error by closing a duplicate consumer
    over the same connection.

### Motivation

Want to remove boost dependencies from CPP Client interface.

### Modifications

Changed boost smart pointers and functions to std.

### Result

No change in the code logic.