Fix bad_weak_ptr when close a ClientConnection during construction

Fixes apache#348

Fixes apache#349

### Motivation

When `close` is called in `ClientConnection`'s constructor,
`shared_from_this()` will be called, which results in a
`std::bad_weak_ptr` error. This error does not happen before
apache#317 because
`shared_from_this()` could only be called when the `producers` or
`consumers` field is not empty.

### Modifications

Modify the 2nd parameter of `ClientConnection::close` to represent if
the construction completes. If not, just set the state to
`Disconnected` and complete the future to the result. Then
`ConnectionPool::getConnectionAsync` will return a future that completes
with the failed result.

In addition, check `authentication_` even for non-TLS URLs. Otherwise,
the null authentication will be used to construct `CommandConnect`.

Add `testInvalidPlugin` and `testTlsConfigError` to verify the changes.

Author: BewareMyPower

lib/ClientConnection.cc

@@ -186,6 +186,11 @@ ClientConnection::ClientConnection(const std::string& logicalAddress, const std:
       clientVersion_(clientVersion),
       pool_(pool) {
     LOG_INFO(cnxString_ << "Create ClientConnection, timeout=" << clientConfiguration.getConnectionTimeout());
+    if (!authentication_) {
+        LOG_ERROR("Invalid authentication plugin");
+        close(ResultAuthenticationError, false);
+        return;
+    }
     if (clientConfiguration.isUseTls()) {
 #if BOOST_VERSION >= 105400
         boost::asio::ssl::context ctx(boost::asio::ssl::context::tlsv12_client);
@@ -214,12 +219,6 @@ ClientConnection::ClientConnection(const std::string& logicalAddress, const std:
             }
         }
 
-        if (!authentication_) {
-            LOG_ERROR("Invalid authentication plugin");
-            close(ResultAuthenticationError, false);
-            return;
-        }
-
         std::string tlsCertificates = clientConfiguration.getTlsCertificateFilePath();
         std::string tlsPrivateKey = clientConfiguration.getTlsPrivateKeyFilePath();
 
@@ -1259,7 +1258,13 @@ void ClientConnection::handleConsumerStatsTimeout(const boost::system::error_cod
     startConsumerStatsTimer(consumerStatsRequests);
 }
 
-void ClientConnection::close(Result result, bool detach) {
+void ClientConnection::close(Result result, bool constructCompleted) {
+    if (!constructCompleted) {
+        state_ = Disconnected;
+        connectPromise_.setFailed(result);
+        return;
+    }
+
     Lock lock(mutex_);
     if (isClosed()) {
         return;
@@ -1319,9 +1324,7 @@ void ClientConnection::close(Result result, bool detach) {
         LOG_INFO(cnxString_ << "Connection disconnected (refCnt: " << refCount << ")");
     }
     // Remove the connection from the pool before completing any promise
-    if (detach) {
-        pool_.remove(logicalAddress_, this);  // trigger the destructor
-    }
+    pool_.remove(logicalAddress_, this);
 
     auto self = shared_from_this();
     for (ProducersMap::iterator it = producers.begin(); it != producers.end(); ++it) {

lib/ClientConnection.h

@@ -146,13 +146,9 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
      * Close the connection.
      *
      * @param result all pending futures will complete with this result
-     * @param detach remove it from the pool if it's true
-     *
-     * `detach` should only be false when:
-     * 1. Before the connection is put into the pool, i.e. during the construction.
-     * 2. When the connection pool is closed
+     * @param constructCompleted whether the construction is completed
      */
-    void close(Result result = ResultConnectError, bool detach = true);
+    void close(Result result = ResultConnectError, bool constructCompleted = true);
 
     bool isClosed() const;
 

tests/AuthPluginTest.cc

@@ -581,3 +581,19 @@ TEST(AuthPluginTest, testOauth2Failure) {
     ASSERT_EQ(client5.createProducer(topic, producer), ResultAuthenticationError);
     client5.close();
 }
+
+TEST(AuthPluginTest, testInvalidPlugin) {
+    Client client("pulsar://localhost:6650", ClientConfiguration{}.setAuth(AuthFactory::create("invalid")));
+    Producer producer;
+    ASSERT_EQ(ResultAuthenticationError, client.createProducer("my-topic", producer));
+    client.close();
+}
+
+TEST(AuthPluginTest, testTlsConfigError) {
+    Client client(serviceUrlTls, ClientConfiguration{}
+                                     .setAuth(AuthTls::create(clientPublicKeyPath, clientPrivateKeyPath))
+                                     .setTlsTrustCertsFilePath("invalid"));
+    Producer producer;
+    ASSERT_EQ(ResultAuthenticationError, client.createProducer("my-topic", producer));
+    client.close();
+}