[improve] add physicalAddress as part of connection pool key (#411)

### Motivation

Context: https://github.com/apache/pulsar/pull/22085/files#r1497008116

Currently, the connection pool key does not include physicalAddress (currently logicalAddress + keySuffix). This can be a problem when the same logicalAddresses are in the migrated(green) cluster. (the connection pool will return the connection to the old(blue) cluster)

### Modifications

Add physicalAddress as part of the connection pool key

Author: heesung-sohn

lib/ClientConnection.cc

@@ -1321,7 +1321,7 @@ void ClientConnection::close(Result result, bool detach) {
     }
     // Remove the connection from the pool before completing any promise
     if (detach) {
-        pool_.remove(logicalAddress_ + "-" + std::to_string(poolIndex_), this);
+        pool_.remove(logicalAddress_, physicalAddress_, poolIndex_, this);
     }
 
     auto self = shared_from_this();

lib/ConnectionPool.cc

@@ -66,6 +66,13 @@ bool ConnectionPool::close() {
     return true;
 }
 
+static const std::string getKey(const std::string& logicalAddress, const std::string& physicalAddress,
+                                size_t keySuffix) {
+    std::stringstream ss;
+    ss << logicalAddress << '-' << physicalAddress << '-' << keySuffix;
+    return ss.str();
+}
+
 Future<Result, ClientConnectionWeakPtr> ConnectionPool::getConnectionAsync(const std::string& logicalAddress,
                                                                            const std::string& physicalAddress,
                                                                            size_t keySuffix) {
@@ -77,9 +84,7 @@ Future<Result, ClientConnectionWeakPtr> ConnectionPool::getConnectionAsync(const
 
     std::unique_lock<std::recursive_mutex> lock(mutex_);
 
-    std::stringstream ss;
-    ss << logicalAddress << '-' << keySuffix;
-    const std::string key = ss.str();
+    auto key = getKey(logicalAddress, physicalAddress, keySuffix);
 
     PoolMap::iterator cnxIt = pool_.find(key);
     if (cnxIt != pool_.end()) {
@@ -127,7 +132,9 @@ Future<Result, ClientConnectionWeakPtr> ConnectionPool::getConnectionAsync(const
     return future;
 }
 
-void ConnectionPool::remove(const std::string& key, ClientConnection* value) {
+void ConnectionPool::remove(const std::string& logicalAddress, const std::string& physicalAddress,
+                            size_t keySuffix, ClientConnection* value) {
+    auto key = getKey(logicalAddress, physicalAddress, keySuffix);
     std::lock_guard<std::recursive_mutex> lock(mutex_);
     auto it = pool_.find(key);
     if (it != pool_.end() && it->second.get() == value) {

lib/ConnectionPool.h

@@ -51,7 +51,8 @@ class PULSAR_PUBLIC ConnectionPool {
      */
     bool close();
 
-    void remove(const std::string& key, ClientConnection* value);
+    void remove(const std::string& logicalAddress, const std::string& physicalAddress, size_t keySuffix,
+                ClientConnection* value);
 
     /**
      * Get a connection from the pool.