Fix ProducerBusy or ConsumerBusy error when configuring multiple brokers per connection (#337)

### Motivation

This is a catch up for apache/pulsar#21144

When a producer or consumer reconnects, a random number will be generated as the key suffix in `ConnectionPool` to create or get the `ClientConnection` object from the pool.

https://github.com/apache/pulsar-client-cpp/blob/81cc562f7b366fad97e1b80c07ef9334a808390d/lib/ConnectionPool.cc#L75

If a new connection is created with the same producer or consumer name to the broker, the broker will respond with a  `ProducerBusy` or `ConsumerBusy` error so that the reconnection will never succeed.

### Modifications

- Add an overload of `ConnectionPool::getConnectionAsync` that accepts an integer parameter as the key suffix. If it's not specified, generate the random number as the suffix. In this method, choose the executor by `key suffix % size`.
- Generate the random number and save it when creating the `HandlerBase` object. When connecting the owner broker of its topic, pass that index so that the reconnection will always reuse the same `ClientConnection` object.

### Verifying this change

`ProducerTest.testReconnectMultiConnectionsPerBroker` is added to protected the change.

Author: BewareMyPower

lib/ClientImpl.cc

@@ -516,7 +516,7 @@ void ClientImpl::handleConsumerCreated(Result result, ConsumerImplBaseWeakPtr co
     }
 }
 
-Future<Result, ClientConnectionPtr> ClientImpl::getConnection(const std::string& topic) {
+Future<Result, ClientConnectionPtr> ClientImpl::getConnection(const std::string& topic, size_t key) {
     Promise<Result, ClientConnectionPtr> promise;
 
     const auto topicNamePtr = TopicName::get(topic);
@@ -528,12 +528,12 @@ Future<Result, ClientConnectionPtr> ClientImpl::getConnection(const std::string&
 
     auto self = shared_from_this();
     lookupServicePtr_->getBroker(*topicNamePtr)
-        .addListener([this, self, promise](Result result, const LookupService::LookupResult& data) {
+        .addListener([this, self, promise, key](Result result, const LookupService::LookupResult& data) {
             if (result != ResultOk) {
                 promise.setFailed(result);
                 return;
             }
-            pool_.getConnectionAsync(data.logicalAddress, data.physicalAddress)
+            pool_.getConnectionAsync(data.logicalAddress, data.physicalAddress, key)
                 .addListener([promise](Result result, const ClientConnectionWeakPtr& weakCnx) {
                     if (result == ResultOk) {
                         auto cnx = weakCnx.lock();

lib/ClientImpl.h

@@ -95,7 +95,7 @@ class ClientImpl : public std::enable_shared_from_this<ClientImpl> {
 
     void getPartitionsForTopicAsync(const std::string& topic, GetPartitionsCallback callback);
 
-    Future<Result, ClientConnectionPtr> getConnection(const std::string& topic);
+    Future<Result, ClientConnectionPtr> getConnection(const std::string& topic, size_t key);
 
     void closeAsync(CloseCallback callback);
     void shutdown();
@@ -123,6 +123,8 @@ class ClientImpl : public std::enable_shared_from_this<ClientImpl> {
 
     std::shared_ptr<std::atomic<uint64_t>> getRequestIdGenerator() const { return requestIdGenerator_; }
 
+    ConnectionPool& getConnectionPool() noexcept { return pool_; }
+
     friend class PulsarFriend;
 
    private:

lib/ConnectionPool.cc

@@ -61,8 +61,9 @@ bool ConnectionPool::close() {
     return true;
 }
 
-Future<Result, ClientConnectionWeakPtr> ConnectionPool::getConnectionAsync(
-    const std::string& logicalAddress, const std::string& physicalAddress) {
+Future<Result, ClientConnectionWeakPtr> ConnectionPool::getConnectionAsync(const std::string& logicalAddress,
+                                                                           const std::string& physicalAddress,
+                                                                           size_t keySuffix) {
     if (closed_) {
         Promise<Result, ClientConnectionWeakPtr> promise;
         promise.setFailed(ResultAlreadyClosed);
@@ -72,7 +73,7 @@ Future<Result, ClientConnectionWeakPtr> ConnectionPool::getConnectionAsync(
     std::unique_lock<std::recursive_mutex> lock(mutex_);
 
     std::stringstream ss;
-    ss << logicalAddress << '-' << randomDistribution_(randomEngine_);
+    ss << logicalAddress << '-' << keySuffix;
     const std::string key = ss.str();
 
     PoolMap::iterator cnxIt = pool_.find(key);
@@ -95,7 +96,7 @@ Future<Result, ClientConnectionWeakPtr> ConnectionPool::getConnectionAsync(
     // No valid or pending connection found in the pool, creating a new one
     ClientConnectionPtr cnx;
     try {
-        cnx.reset(new ClientConnection(logicalAddress, physicalAddress, executorProvider_->get(),
+        cnx.reset(new ClientConnection(logicalAddress, physicalAddress, executorProvider_->get(keySuffix),
                                        clientConfiguration_, authentication_, clientVersion_, *this));
     } catch (const std::runtime_error& e) {
         lock.unlock();

lib/ConnectionPool.h

@@ -65,17 +65,29 @@ class PULSAR_PUBLIC ConnectionPool {
      * a proxy layer. Essentially, the pool is using the logical address as a way to
      * decide whether to reuse a particular connection.
      *
+     * There could be many connections to the same broker, so this pool uses an integer key as the suffix of
+     * the key that represents the connection.
+     *
      * @param logicalAddress the address to use as the broker tag
      * @param physicalAddress the real address where the TCP connection should be made
+     * @param keySuffix the key suffix to choose which connection on the same broker
      * @return a future that will produce the ClientCnx object
      */
     Future<Result, ClientConnectionWeakPtr> getConnectionAsync(const std::string& logicalAddress,
-                                                               const std::string& physicalAddress);
+                                                               const std::string& physicalAddress,
+                                                               size_t keySuffix);
+
+    Future<Result, ClientConnectionWeakPtr> getConnectionAsync(const std::string& logicalAddress,
+                                                               const std::string& physicalAddress) {
+        return getConnectionAsync(logicalAddress, physicalAddress, generateRandomIndex());
+    }
 
     Future<Result, ClientConnectionWeakPtr> getConnectionAsync(const std::string& address) {
         return getConnectionAsync(address, address);
     }
 
+    size_t generateRandomIndex() { return randomDistribution_(randomEngine_); }
+
    private:
     ClientConfiguration clientConfiguration_;
     ExecutorServiceProviderPtr executorProvider_;

lib/ExecutorService.cc

@@ -133,10 +133,10 @@ void ExecutorService::postWork(std::function<void(void)> task) { io_service_.pos
 ExecutorServiceProvider::ExecutorServiceProvider(int nthreads)
     : executors_(nthreads), executorIdx_(0), mutex_() {}
 
-ExecutorServicePtr ExecutorServiceProvider::get() {
+ExecutorServicePtr ExecutorServiceProvider::get(size_t idx) {
+    idx %= executors_.size();
     Lock lock(mutex_);
 
-    int idx = executorIdx_++ % executors_.size();
     if (!executors_[idx]) {
         executors_[idx] = ExecutorService::create();
     }

lib/ExecutorService.h

@@ -88,15 +88,17 @@ class PULSAR_PUBLIC ExecutorServiceProvider {
    public:
     explicit ExecutorServiceProvider(int nthreads);
 
-    ExecutorServicePtr get();
+    ExecutorServicePtr get() { return get(executorIdx_++); }
+
+    ExecutorServicePtr get(size_t index);
 
     // See TimeoutProcessor for the semantics of the parameter.
     void close(long timeoutMs = 3000);
 
    private:
     typedef std::vector<ExecutorServicePtr> ExecutorList;
     ExecutorList executors_;
-    int executorIdx_;
+    std::atomic_size_t executorIdx_;
     std::mutex mutex_;
     typedef std::unique_lock<std::mutex> Lock;
 };

lib/HandlerBase.cc

@@ -32,6 +32,7 @@ namespace pulsar {
 HandlerBase::HandlerBase(const ClientImplPtr& client, const std::string& topic, const Backoff& backoff)
     : topic_(std::make_shared<std::string>(topic)),
       client_(client),
+      connectionKeySuffix_(client->getConnectionPool().generateRandomIndex()),
       executor_(client->getIOExecutorProvider()->get()),
       mutex_(),
       creationTimestamp_(TimeUtils::now()),
@@ -88,7 +89,8 @@ void HandlerBase::grabCnx() {
         return;
     }
     auto self = shared_from_this();
-    client->getConnection(topic()).addListener([this, self](Result result, const ClientConnectionPtr& cnx) {
+    auto cnxFuture = client->getConnection(topic(), connectionKeySuffix_);
+    cnxFuture.addListener([this, self](Result result, const ClientConnectionPtr& cnx) {
         if (result == ResultOk) {
             LOG_DEBUG(getName() << "Connected to broker: " << cnx->cnxString());
             connectionOpened(cnx).addListener([this, self](Result result, bool) {

lib/HandlerBase.h

@@ -99,6 +99,7 @@ class HandlerBase : public std::enable_shared_from_this<HandlerBase> {
 
    protected:
     ClientImplWeakPtr client_;
+    const size_t connectionKeySuffix_;
     ExecutorServicePtr executor_;
     mutable std::mutex mutex_;
     std::mutex pendingReceiveMutex_;

lib/ProducerImpl.cc

@@ -966,7 +966,7 @@ bool ProducerImpl::encryptMessage(proto::MessageMetadata& metadata, SharedBuffer
 }
 
 void ProducerImpl::disconnectProducer() {
-    LOG_DEBUG("Broker notification of Closed producer: " << producerId_);
+    LOG_INFO("Broker notification of Closed producer: " << producerId_);
     resetCnx();
     scheduleReconnection();
 }

tests/ProducerTest.cc

@@ -618,4 +618,19 @@ TEST(ProducerTest, testNoDeadlockWhenClosingPartitionedProducerAfterPartitionsUp
     client.close();
 }
 
+TEST(ProducerTest, testReconnectMultiConnectionsPerBroker) {
+    ClientConfiguration conf;
+    conf.setConnectionsPerBroker(10);
+
+    Client client(serviceUrl, conf);
+    Producer producer;
+    ASSERT_EQ(ResultOk, client.createProducer("producer-test-reconnect-twice", producer));
+
+    for (int i = 0; i < 5; i++) {
+        ASSERT_TRUE(PulsarFriend::reconnect(producer)) << "i: " << i;
+    }
+
+    client.close();
+}
+
 INSTANTIATE_TEST_CASE_P(Pulsar, ProducerTest, ::testing::Values(true, false));