[C++] Catch exception thrown by remote_endpoint (apache#8486)

### Motivation

Boost asio `socket::remote_point` may throw `boost::system::system_error` on failure. If the C++ client library was compiled with some low version boost, like 1.54, even if `async_connect` success, the server could still be unreachable and an exception would be thrown by `socket_->remote_endpoint()`.

### Modifications

- Catch the exception for  `socket_->remote_endpoint()`
- Add tests for when client connects to a unreachable service url
- Fix header dependency because some boost *.hpp file doesn't include assert.h

### Verifying this change

- [ ] Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

  - Run `testServerConnectError`.

pulsar-client-cpp/lib/BlockingQueue.h

@@ -19,6 +19,7 @@
 #ifndef LIB_BLOCKINGQUEUE_H_
 #define LIB_BLOCKINGQUEUE_H_
 
+#include <assert.h>
 #include <mutex>
 #include <condition_variable>
 #include <boost/circular_buffer.hpp>

pulsar-client-cpp/lib/ClientConnection.cc

@@ -340,9 +340,15 @@ void ClientConnection::handleTcpConnected(const boost::system::error_code& err,
                                           tcp::resolver::iterator endpointIterator) {
     if (!err) {
         std::stringstream cnxStringStream;
-        cnxStringStream << "[" << socket_->local_endpoint() << " -> " << socket_->remote_endpoint() << "] ";
-        cnxString_ = cnxStringStream.str();
-
+        try {
+            cnxStringStream << "[" << socket_->local_endpoint() << " -> " << socket_->remote_endpoint()
+                            << "] ";
+            cnxString_ = cnxStringStream.str();
+        } catch (const boost::system::system_error& e) {
+            LOG_ERROR("Failed to get endpoint: " << e.what());
+            close();
+            return;
+        }
         if (logicalAddress_ == physicalAddress_) {
             LOG_INFO(cnxString_ << "Connected to broker");
         } else {

pulsar-client-cpp/tests/ClientTest.cc

@@ -73,3 +73,16 @@ TEST(ClientTest, testSwHwChecksum) {
     ASSERT_EQ(hwIncrementalChecksum, hwDoubleChecksum);
     ASSERT_EQ(hwIncrementalChecksum, swIncrementalChecksum);
 }
+
+TEST(ClientTest, testServerConnectError) {
+    const std::string topic = "test-server-connect-error";
+    Client client("pulsar://localhost:65535");
+    Producer producer;
+    ASSERT_EQ(ResultConnectError, client.createProducer(topic, producer));
+    Consumer consumer;
+    ASSERT_EQ(ResultConnectError, client.subscribe(topic, "sub", consumer));
+    Reader reader;
+    ReaderConfiguration readerConf;
+    ASSERT_EQ(ResultConnectError, client.createReader(topic, MessageId::earliest(), readerConf, reader));
+    client.close();
+}