Revert unrelated changes

Author: BewareMyPower

lib/ClientConnection.cc

@@ -278,7 +278,7 @@ ClientConnection::~ClientConnection() {
 void ClientConnection::handlePulsarConnected(const proto::CommandConnected& cmdConnected) {
     if (!cmdConnected.has_server_version()) {
         LOG_ERROR(cnxString_ << "Server version is not set");
-        close(ResultUnknownError);
+        close();
         return;
     }
 
@@ -451,7 +451,7 @@ void ClientConnection::handleTcpConnected(const ASIO_ERROR& err, tcp::resolver::
                 Url service_url;
                 if (!Url::parse(physicalAddress_, service_url)) {
                     LOG_ERROR(cnxString_ << "Invalid Url, unable to parse: " << err << " " << err.message());
-                    close(ResultInvalidUrl);
+                    close();
                     return;
                 }
             }
@@ -489,7 +489,12 @@ void ClientConnection::handleTcpConnected(const ASIO_ERROR& err, tcp::resolver::
                 }
             });
         } else {
-            close(ResultRetryable);
+            if (err == ASIO::error::operation_aborted) {
+                // TCP connect timeout, which is not retryable
+                close();
+            } else {
+                close(ResultRetryable);
+            }
         }
     } else {
         LOG_ERROR(cnxString_ << "Failed to establish connection: " << err.message());
@@ -499,8 +504,8 @@ void ClientConnection::handleTcpConnected(const ASIO_ERROR& err, tcp::resolver::
 
 void ClientConnection::handleHandshake(const ASIO_ERROR& err) {
     if (err) {
-        LOG_WARN(cnxString_ << "Handshake failed: " << err.message());
-        close(ResultRetryable);
+        LOG_ERROR(cnxString_ << "Handshake failed: " << err.message());
+        close();
         return;
     }
 
@@ -527,7 +532,7 @@ void ClientConnection::handleSentPulsarConnect(const ASIO_ERROR& err, const Shar
     }
     if (err) {
         LOG_ERROR(cnxString_ << "Failed to establish connection: " << err.message());
-        close(ResultRetryable);
+        close();
         return;
     }
 
@@ -562,14 +567,14 @@ void ClientConnection::tcpConnectAsync() {
     std::string hostUrl = isSniProxy_ ? proxyServiceUrl_ : physicalAddress_;
     if (!Url::parse(hostUrl, service_url)) {
         LOG_ERROR(cnxString_ << "Invalid Url, unable to parse: " << err << " " << err.message());
-        close(ResultInvalidUrl);
+        close();
         return;
     }
 
     if (service_url.protocol() != "pulsar" && service_url.protocol() != "pulsar+ssl") {
         LOG_ERROR(cnxString_ << "Invalid Url protocol '" << service_url.protocol()
                              << "'. Valid values are 'pulsar' and 'pulsar+ssl'");
-        close(ResultInvalidUrl);
+        close();
         return;
     }
 
@@ -588,7 +593,7 @@ void ClientConnection::handleResolve(const ASIO_ERROR& err, tcp::resolver::itera
     if (err) {
         std::string hostUrl = isSniProxy_ ? cnxString_ : proxyServiceUrl_;
         LOG_ERROR(hostUrl << "Resolve error: " << err << " : " << err.message());
-        close(ResultConnectError);
+        close();
         return;
     }
 
@@ -624,8 +629,8 @@ void ClientConnection::handleResolve(const ASIO_ERROR& err, tcp::resolver::itera
             }
         });
     } else {
-        LOG_ERROR(cnxString_ << "No IP address found");
-        close(ResultConnectError);
+        LOG_WARN(cnxString_ << "No IP address found");
+        close();
         return;
     }
 }
@@ -880,7 +885,7 @@ void ClientConnection::handleIncomingCommand(BaseCommand& incomingCmd) {
             // Handle Pulsar Connected
             if (incomingCmd.type() != BaseCommand::CONNECTED) {
                 // Wrong cmd
-                close(ResultRetryable);
+                close();
             } else {
                 handlePulsarConnected(incomingCmd.connected());
             }

lib/ClientConnection.h

@@ -157,7 +157,7 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
      *
      * `detach` should only be false when the connection pool is closed.
      */
-    void close(Result result, bool detach = true);
+    void close(Result result = ResultConnectError, bool detach = true);
 
     bool isClosed() const;
 

tests/BasicEndToEndTest.cc

@@ -1245,7 +1245,7 @@ TEST(BasicEndToEndTest, testHandlerReconnectionLogic) {
                 std::this_thread::sleep_for(std::chrono::seconds(1));
             } while (!clientConnectionPtr);
             oldConnections.push_back(clientConnectionPtr);
-            clientConnectionPtr->close(ResultConnectError);
+            clientConnectionPtr->close();
         }
         LOG_INFO("checking message " << i);
         ASSERT_EQ(producer.send(msg), ResultOk);
@@ -1264,7 +1264,7 @@ TEST(BasicEndToEndTest, testHandlerReconnectionLogic) {
     ClientConnectionWeakPtr clientConnectionWeakPtr = PulsarFriend::getClientConnection(cImpl);
     ClientConnectionPtr clientConnectionPtr = clientConnectionWeakPtr.lock();
     oldConnections.push_back(clientConnectionPtr);
-    clientConnectionPtr->close(ResultConnectError);
+    clientConnectionPtr->close();
 
     while (consumer.receive(msg, 30000) == ResultOk) {
         consumer.acknowledge(msg);
@@ -1316,7 +1316,7 @@ void testHandlerReconnectionPartitionProducers(bool lazyStartPartitionedProducer
                 std::this_thread::sleep_for(std::chrono::seconds(1));
             } while (!clientConnectionPtr);
             oldConnections.push_back(clientConnectionPtr);
-            clientConnectionPtr->close(ResultConnectError);
+            clientConnectionPtr->close();
         }
         ASSERT_EQ(producer.send(msg), ResultOk);
     }

tests/ClientTest.cc

@@ -412,7 +412,7 @@ TEST(ClientTest, testConnectionClose) {
             auto executor = PulsarFriend::getExecutor(*cnx);
             // Simulate the close() happens in the event loop
             executor->postWork([cnx, &client, numConnections] {
-                cnx->close(ResultConnectError);
+                cnx->close();
                 ASSERT_EQ(PulsarFriend::getConnections(client).size(), numConnections - 1);
                 LOG_INFO("Connection refcnt: " << cnx.use_count() << " after close");
             });