Revert "Use the same connection when producer or consumer reconnect"

BewareMyPower · BewareMyPower · commit bfb48a03eda0 · 2023-11-06T14:18:36.000+08:00
This reverts commit 768c75a.
diff --git a/lib/BinaryProtoLookupService.cc b/lib/BinaryProtoLookupService.cc
@@ -47,10 +47,9 @@ auto BinaryProtoLookupService::findBroker(const std::string& address, bool autho
     }
 
     // NOTE: we can use move capture for topic since C++14
-    cnxPool_.getRandomConnectionAsync(address).addListener([this, promise, topic, address, authoritative,
-                                                            redirectCount](
-                                                               Result result,
-                                                               const ClientConnectionWeakPtr& weakCnx) {
+    cnxPool_.getConnectionAsync(address).addListener([this, promise, topic, address, authoritative,
+                                                      redirectCount](Result result,
+                                                                     const ClientConnectionWeakPtr& weakCnx) {
         if (result != ResultOk) {
             promise->setFailed(result);
             return;
@@ -110,7 +109,7 @@ Future<Result, LookupDataResultPtr> BinaryProtoLookupService::getPartitionMetada
     }
     std::string lookupName = topicName->toString();
     const auto address = serviceNameResolver_.resolveHost();
-    cnxPool_.getRandomConnectionAsync(address, address)
+    cnxPool_.getConnectionAsync(address, address)
         .addListener(std::bind(&BinaryProtoLookupService::sendPartitionMetadataLookupRequest, this,
                                lookupName, std::placeholders::_1, std::placeholders::_2, promise));
     return promise->getFuture();
@@ -163,7 +162,7 @@ Future<Result, NamespaceTopicsPtr> BinaryProtoLookupService::getTopicsOfNamespac
         return promise->getFuture();
     }
     std::string namespaceName = nsName->toString();
-    cnxPool_.getRandomConnectionAsync(serviceNameResolver_.resolveHost())
+    cnxPool_.getConnectionAsync(serviceNameResolver_.resolveHost())
         .addListener(std::bind(&BinaryProtoLookupService::sendGetTopicsOfNamespaceRequest, this,
                                namespaceName, mode, std::placeholders::_1, std::placeholders::_2, promise));
     return promise->getFuture();
@@ -177,7 +176,7 @@ Future<Result, SchemaInfo> BinaryProtoLookupService::getSchema(const TopicNamePt
         promise->setFailed(ResultInvalidTopicName);
         return promise->getFuture();
     }
-    cnxPool_.getRandomConnectionAsync(serviceNameResolver_.resolveHost())
+    cnxPool_.getConnectionAsync(serviceNameResolver_.resolveHost())
         .addListener(std::bind(&BinaryProtoLookupService::sendGetSchemaRequest, this, topicName->toString(),
                                version, std::placeholders::_1, std::placeholders::_2, promise));
 
diff --git a/lib/ClientImpl.cc b/lib/ClientImpl.cc
@@ -516,7 +516,7 @@ void ClientImpl::handleConsumerCreated(Result result, ConsumerImplBaseWeakPtr co
     }
 }
 
-Future<Result, ClientConnectionPtr> ClientImpl::getConnection(const std::string& topic, int key) {
+Future<Result, ClientConnectionPtr> ClientImpl::getConnection(const std::string& topic) {
     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, key](Result result, const LookupService::LookupResult& data) {
+        .addListener([this, self, promise](Result result, const LookupService::LookupResult& data) {
             if (result != ResultOk) {
                 promise.setFailed(result);
                 return;
             }
-            pool_.getConnectionAsync(data.logicalAddress, data.physicalAddress, key)
+            pool_.getConnectionAsync(data.logicalAddress, data.physicalAddress)
                 .addListener([promise](Result result, const ClientConnectionWeakPtr& weakCnx) {
                     if (result == ResultOk) {
                         auto cnx = weakCnx.lock();
diff --git a/lib/ClientImpl.h b/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, int key);
+    Future<Result, ClientConnectionPtr> getConnection(const std::string& topic);
 
     void closeAsync(CloseCallback callback);
     void shutdown();
@@ -123,8 +123,6 @@ 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:
diff --git a/lib/ConnectionPool.cc b/lib/ConnectionPool.cc
@@ -61,9 +61,8 @@ bool ConnectionPool::close() {
     return true;
 }
 
-Future<Result, ClientConnectionWeakPtr> ConnectionPool::getConnectionAsync(const std::string& logicalAddress,
-                                                                           const std::string& physicalAddress,
-                                                                           int keySuffix) {
+Future<Result, ClientConnectionWeakPtr> ConnectionPool::getConnectionAsync(
+    const std::string& logicalAddress, const std::string& physicalAddress) {
     if (closed_) {
         Promise<Result, ClientConnectionWeakPtr> promise;
         promise.setFailed(ResultAlreadyClosed);
@@ -73,7 +72,7 @@ Future<Result, ClientConnectionWeakPtr> ConnectionPool::getConnectionAsync(const
     std::unique_lock<std::recursive_mutex> lock(mutex_);
 
     std::stringstream ss;
-    ss << logicalAddress << '-' << keySuffix;
+    ss << logicalAddress << '-' << randomDistribution_(randomEngine_);
     const std::string key = ss.str();
 
     PoolMap::iterator cnxIt = pool_.find(key);
@@ -96,8 +95,7 @@ Future<Result, ClientConnectionWeakPtr> ConnectionPool::getConnectionAsync(const
     // No valid or pending connection found in the pool, creating a new one
     ClientConnectionPtr cnx;
     try {
-        size_t index = executorIndex_++;
-        cnx.reset(new ClientConnection(logicalAddress, physicalAddress, executorProvider_->get(index),
+        cnx.reset(new ClientConnection(logicalAddress, physicalAddress, executorProvider_->get(),
                                        clientConfiguration_, authentication_, clientVersion_, *this));
     } catch (const std::runtime_error& e) {
         lock.unlock();
diff --git a/lib/ConnectionPool.h b/lib/ConnectionPool.h
@@ -65,27 +65,15 @@ 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,
-                                                               int keySuffix);
-
-    int generateRandomIndex() { return randomDistribution_(randomEngine_); }
-
-    Future<Result, ClientConnectionWeakPtr> getRandomConnectionAsync(const std::string& logicalAddress,
-                                                                     const std::string& physicalAddress) {
-        return getConnectionAsync(logicalAddress, physicalAddress, generateRandomIndex());
-    }
+                                                               const std::string& physicalAddress);
 
-    Future<Result, ClientConnectionWeakPtr> getRandomConnectionAsync(const std::string& address) {
-        return getRandomConnectionAsync(address, address);
+    Future<Result, ClientConnectionWeakPtr> getConnectionAsync(const std::string& address) {
+        return getConnectionAsync(address, address);
     }
 
    private:
@@ -97,8 +85,6 @@ class PULSAR_PUBLIC ConnectionPool {
     const std::string clientVersion_;
     mutable std::recursive_mutex mutex_;
     std::atomic_bool closed_{false};
-    // Use a separated index so that connections will be distributed uniformly across the executors
-    std::atomic_size_t executorIndex_{0};
 
     std::uniform_int_distribution<> randomDistribution_;
     std::mt19937 randomEngine_;
diff --git a/lib/ExecutorService.cc b/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(size_t idx) {
-    idx %= executors_.size();
+ExecutorServicePtr ExecutorServiceProvider::get() {
     Lock lock(mutex_);
 
+    int idx = executorIdx_++ % executors_.size();
     if (!executors_[idx]) {
         executors_[idx] = ExecutorService::create();
     }
diff --git a/lib/ExecutorService.h b/lib/ExecutorService.h
@@ -88,17 +88,15 @@ class PULSAR_PUBLIC ExecutorServiceProvider {
    public:
     explicit ExecutorServiceProvider(int nthreads);
 
-    ExecutorServicePtr get() { return get(executorIdx_++); }
-
-    ExecutorServicePtr get(size_t index);
+    ExecutorServicePtr get();
 
     // See TimeoutProcessor for the semantics of the parameter.
     void close(long timeoutMs = 3000);
 
    private:
     typedef std::vector<ExecutorServicePtr> ExecutorList;
     ExecutorList executors_;
-    std::atomic_size_t executorIdx_;
+    int executorIdx_;
     std::mutex mutex_;
     typedef std::unique_lock<std::mutex> Lock;
 };
diff --git a/lib/HandlerBase.cc b/lib/HandlerBase.cc
@@ -32,7 +32,6 @@ 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()),
@@ -89,23 +88,22 @@ void HandlerBase::grabCnx() {
         return;
     }
     auto self = shared_from_this();
-    client->getConnection(topic(), connectionKeySuffix_)
-        .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) {
-                    // Do not use bool, only Result.
-                    reconnectionPending_ = false;
-                    if (isResultRetryable(result)) {
-                        scheduleReconnection();
-                    }
-                });
-            } else {
-                connectionFailed(result);
+    client->getConnection(topic()).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) {
+                // Do not use bool, only Result.
                 reconnectionPending_ = false;
-                scheduleReconnection();
-            }
-        });
+                if (isResultRetryable(result)) {
+                    scheduleReconnection();
+                }
+            });
+        } else {
+            connectionFailed(result);
+            reconnectionPending_ = false;
+            scheduleReconnection();
+        }
+    });
 }
 
 void HandlerBase::handleDisconnection(Result result, const ClientConnectionPtr& cnx) {
diff --git a/lib/HandlerBase.h b/lib/HandlerBase.h
@@ -99,7 +99,6 @@ class HandlerBase : public std::enable_shared_from_this<HandlerBase> {
 
    protected:
     ClientImplWeakPtr client_;
-    const int connectionKeySuffix_;
     ExecutorServicePtr executor_;
     mutable std::mutex mutex_;
     std::mutex pendingReceiveMutex_;
diff --git a/lib/ProducerImpl.cc b/lib/ProducerImpl.cc
@@ -966,7 +966,7 @@ bool ProducerImpl::encryptMessage(proto::MessageMetadata& metadata, SharedBuffer
 }
 
 void ProducerImpl::disconnectProducer() {
-    LOG_INFO("Broker notification of Closed producer: " << producerId_);
+    LOG_DEBUG("Broker notification of Closed producer: " << producerId_);
     resetCnx();
     scheduleReconnection();
 }

Original file line number	Diff line number	Diff line change
`@@ -966,7 +966,7 @@ bool ProducerImpl::encryptMessage(proto::MessageMetadata& metadata, SharedBuffer`
`966`	`966`	`}`
`967`	`967`
`968`	`968`	`void ProducerImpl::disconnectProducer() {`
`969`		`- LOG_INFO("Broker notification of Closed producer: " << producerId_);`
	`969`	`+ LOG_DEBUG("Broker notification of Closed producer: " << producerId_);`
`970`	`970`	`resetCnx();`
`971`	`971`	`scheduleReconnection();`
`972`	`972`	`}`