Fix creating producer or consumer is not retried for connection failure

BewareMyPower · BewareMyPower · commit c4d147397b54 · 2024-02-04T20:26:25.000+08:00
Fixes #391 ### Motivation When `connectionFailed` is called, no matter if the result is retryable the creation of producer or consumer will fail without retry. ### Modifications Check if the result is retryable in `connectionFailed` for `ProducerImpl` and `ConsumerImpl` and only fail for non-retryable errors or the timeout error. Register another timer in `HandlerBase` to propagate the timeout error to `connectionFailed`. Add `testRetryUntilSucceed`, `testRetryTimeout`, `testNoRetry` to verify client could retry according to the result returned by `ClientImpl::getConnection`. On the other handle, check all `close()` calls in `ClientConnection` and pass the correct result. For example, a handshake error should be retryable so we must call `close(ResultRetryable)`.
diff --git a/lib/ClientConnection.cc b/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();
+        close(ResultUnknownError);
         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();
+                    close(ResultInvalidUrl);
                     return;
                 }
             }
@@ -489,12 +489,7 @@ void ClientConnection::handleTcpConnected(const ASIO_ERROR& err, tcp::resolver::
                 }
             });
         } else {
-            if (err == ASIO::error::operation_aborted) {
-                // TCP connect timeout, which is not retryable
-                close();
-            } else {
-                close(ResultRetryable);
-            }
+            close(ResultRetryable);
         }
     } else {
         LOG_ERROR(cnxString_ << "Failed to establish connection: " << err.message());
@@ -504,8 +499,8 @@ void ClientConnection::handleTcpConnected(const ASIO_ERROR& err, tcp::resolver::
 
 void ClientConnection::handleHandshake(const ASIO_ERROR& err) {
     if (err) {
-        LOG_ERROR(cnxString_ << "Handshake failed: " << err.message());
-        close();
+        LOG_WARN(cnxString_ << "Handshake failed: " << err.message());
+        close(ResultRetryable);
         return;
     }
 
@@ -532,7 +527,7 @@ void ClientConnection::handleSentPulsarConnect(const ASIO_ERROR& err, const Shar
     }
     if (err) {
         LOG_ERROR(cnxString_ << "Failed to establish connection: " << err.message());
-        close();
+        close(ResultRetryable);
         return;
     }
 
@@ -546,7 +541,7 @@ void ClientConnection::handleSentAuthResponse(const ASIO_ERROR& err, const Share
     }
     if (err) {
         LOG_WARN(cnxString_ << "Failed to send auth response: " << err.message());
-        close();
+        close(ResultRetryable);
         return;
     }
 }
@@ -567,14 +562,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();
+        close(ResultInvalidUrl);
         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();
+        close(ResultInvalidUrl);
         return;
     }
 
@@ -593,7 +588,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();
+        close(ResultConnectError);
         return;
     }
 
@@ -629,8 +624,8 @@ void ClientConnection::handleResolve(const ASIO_ERROR& err, tcp::resolver::itera
             }
         });
     } else {
-        LOG_WARN(cnxString_ << "No IP address found");
-        close();
+        LOG_ERROR(cnxString_ << "No IP address found");
+        close(ResultConnectError);
         return;
     }
 }
@@ -885,7 +880,7 @@ void ClientConnection::handleIncomingCommand(BaseCommand& incomingCmd) {
             // Handle Pulsar Connected
             if (incomingCmd.type() != BaseCommand::CONNECTED) {
                 // Wrong cmd
-                close();
+                close(ResultRetryable);
             } else {
                 handlePulsarConnected(incomingCmd.connected());
             }
diff --git a/lib/ClientConnection.h b/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 = ResultConnectError, bool detach = true);
+    void close(Result result, bool detach = true);
 
     bool isClosed() const;
 
diff --git a/lib/ClientImpl.cc b/lib/ClientImpl.cc
@@ -517,7 +517,7 @@ void ClientImpl::handleConsumerCreated(Result result, ConsumerImplBaseWeakPtr co
     }
 }
 
-Future<Result, ClientConnectionPtr> ClientImpl::getConnection(const std::string& topic, size_t key) {
+GetConnectionFuture ClientImpl::getConnection(const std::string& topic, size_t key) {
     Promise<Result, ClientConnectionPtr> promise;
 
     const auto topicNamePtr = TopicName::get(topic);
@@ -562,7 +562,7 @@ const std::string& ClientImpl::getPhysicalAddress(const std::string& logicalAddr
     }
 }
 
-Future<Result, ClientConnectionPtr> ClientImpl::connect(const std::string& logicalAddress, size_t key) {
+GetConnectionFuture ClientImpl::connect(const std::string& logicalAddress, size_t key) {
     const auto& physicalAddress = getPhysicalAddress(logicalAddress);
     Promise<Result, ClientConnectionPtr> promise;
     pool_.getConnectionAsync(logicalAddress, physicalAddress, key)
diff --git a/lib/ClientImpl.h b/lib/ClientImpl.h
@@ -63,13 +63,14 @@ class TopicName;
 using TopicNamePtr = std::shared_ptr<TopicName>;
 
 using NamespaceTopicsPtr = std::shared_ptr<std::vector<std::string>>;
+using GetConnectionFuture = Future<Result, ClientConnectionPtr>;
 
 std::string generateRandomName();
 
 class ClientImpl : public std::enable_shared_from_this<ClientImpl> {
    public:
     ClientImpl(const std::string& serviceUrl, const ClientConfiguration& clientConfiguration);
-    ~ClientImpl();
+    virtual ~ClientImpl();
 
     /**
      * @param autoDownloadSchema When it is true, Before creating a producer, it will try to get the schema
@@ -95,9 +96,10 @@ 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, size_t key);
+    // Use virtual method to test
+    virtual GetConnectionFuture getConnection(const std::string& topic, size_t key);
 
-    Future<Result, ClientConnectionPtr> connect(const std::string& logicalAddress, size_t key);
+    GetConnectionFuture connect(const std::string& logicalAddress, size_t key);
 
     void closeAsync(CloseCallback callback);
     void shutdown();
diff --git a/lib/ConsumerImpl.cc b/lib/ConsumerImpl.cc
@@ -277,7 +277,7 @@ void ConsumerImpl::connectionFailed(Result result) {
     // Keep a reference to ensure object is kept alive
     auto ptr = get_shared_this_ptr();
 
-    if (consumerCreatedPromise_.setFailed(result)) {
+    if (!isResultRetryable(result) && consumerCreatedPromise_.setFailed(result)) {
         state_ = Failed;
     }
 }
diff --git a/lib/Future.h b/lib/Future.h
@@ -116,6 +116,8 @@ class Future {
 
     Result get(Type &result) { return state_->get(result); }
 
+    static Future<Result, Type> failed(Result result);
+
    private:
     InternalStatePtr<Result, Type> state_;
 
@@ -144,6 +146,13 @@ class Promise {
     InternalStatePtr<Result, Type> state_;
 };
 
+template <typename Result, typename Type>
+inline Future<Result, Type> Future<Result, Type>::failed(Result result) {
+    Promise<Result, Type> promise;
+    promise.setFailed(result);
+    return promise.getFuture();
+}
+
 }  // namespace pulsar
 
 #endif
diff --git a/lib/HandlerBase.cc b/lib/HandlerBase.cc
@@ -42,16 +42,31 @@ HandlerBase::HandlerBase(const ClientImplPtr& client, const std::string& topic,
       backoff_(backoff),
       epoch_(0),
       timer_(executor_->createDeadlineTimer()),
+      creationTimer_(executor_->createDeadlineTimer()),
       reconnectionPending_(false) {}
 
-HandlerBase::~HandlerBase() { timer_->cancel(); }
+HandlerBase::~HandlerBase() {
+    ASIO_ERROR ignored;
+    timer_->cancel(ignored);
+    creationTimer_->cancel(ignored);
+}
 
 void HandlerBase::start() {
     // guard against concurrent state changes such as closing
     State state = NotStarted;
     if (state_.compare_exchange_strong(state, Pending)) {
         grabCnx();
     }
+    creationTimer_->expires_from_now(operationTimeut_);
+    std::weak_ptr<HandlerBase> weakSelf{shared_from_this()};
+    creationTimer_->async_wait([this, weakSelf](const ASIO_ERROR& error) {
+        auto self = weakSelf.lock();
+        if (self && !error) {
+            connectionFailed(ResultTimeout);
+            ASIO_ERROR ignored;
+            timer_->cancel(ignored);
+        }
+    });
 }
 
 ClientConnectionWeakPtr HandlerBase::getCnx() const {
@@ -96,7 +111,7 @@ void HandlerBase::grabCnx(const boost::optional<std::string>& assignedBrokerUrl)
     ClientImplPtr client = client_.lock();
     if (!client) {
         LOG_WARN(getName() << "Client is invalid when calling grabCnx()");
-        connectionFailed(ResultConnectError);
+        connectionFailed(ResultAlreadyClosed);
         reconnectionPending_ = false;
         return;
     }
diff --git a/lib/HandlerBase.h b/lib/HandlerBase.h
@@ -140,6 +140,7 @@ class HandlerBase : public std::enable_shared_from_this<HandlerBase> {
 
    private:
     DeadlineTimerPtr timer_;
+    DeadlineTimerPtr creationTimer_;
 
     mutable std::mutex connectionMutex_;
     std::atomic<bool> reconnectionPending_;
diff --git a/lib/ProducerImpl.cc b/lib/ProducerImpl.cc
@@ -176,7 +176,7 @@ void ProducerImpl::connectionFailed(Result result) {
         // if producers are lazy, then they should always try to restart
         // so don't change the state and allow reconnections
         return;
-    } else if (producerCreatedPromise_.setFailed(result)) {
+    } else if (!isResultRetryable(result) && producerCreatedPromise_.setFailed(result)) {
         state_ = Failed;
     }
 }
diff --git a/tests/BasicEndToEndTest.cc b/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();
+            clientConnectionPtr->close(ResultConnectError);
         }
         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();
+    clientConnectionPtr->close(ResultConnectError);
 
     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();
+            clientConnectionPtr->close(ResultConnectError);
         }
         ASSERT_EQ(producer.send(msg), ResultOk);
     }
diff --git a/tests/ClientTest.cc b/tests/ClientTest.cc
@@ -25,6 +25,7 @@
 #include <future>
 #include <sstream>
 
+#include "MockClientImpl.h"
 #include "PulsarAdminHelper.h"
 #include "PulsarFriend.h"
 #include "WaitUtils.h"
@@ -36,6 +37,7 @@
 DECLARE_LOG_OBJECT()
 
 using namespace pulsar;
+using testing::AtLeast;
 
 static std::string lookupUrl = "pulsar://localhost:6650";
 static std::string adminUrl = "http://localhost:8080/";
@@ -410,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();
+                cnx->close(ResultConnectError);
                 ASSERT_EQ(PulsarFriend::getConnections(client).size(), numConnections - 1);
                 LOG_INFO("Connection refcnt: " << cnx.use_count() << " after close");
             });
@@ -434,3 +436,68 @@ TEST(ClientTest, testConnectionClose) {
         client.close();
     }
 }
+
+TEST(ClientTest, testRetryUntilSucceed) {
+    auto clientImpl = std::make_shared<MockClientImpl>(lookupUrl);
+    constexpr int kFailCount = 3;
+    EXPECT_CALL(*clientImpl, getConnection).Times((kFailCount + 1) * 2);
+    std::atomic_int count{0};
+    ON_CALL(*clientImpl, getConnection)
+        .WillByDefault([&clientImpl, &count](const std::string &topic, size_t index) {
+            if (count++ < kFailCount) {
+                return GetConnectionFuture::failed(ResultRetryable);
+            }
+            return clientImpl->getConnectionReal(topic, index);
+        });
+
+    auto topic = "client-test-retry-until-succeed";
+    ASSERT_EQ(ResultOk, clientImpl->createProducer(topic).result);
+    count = 0;
+    ASSERT_EQ(ResultOk, clientImpl->subscribe(topic).result);
+    ASSERT_EQ(ResultOk, clientImpl->close());
+}
+
+TEST(ClientTest, testRetryTimeout) {
+    auto clientImpl =
+        std::make_shared<MockClientImpl>(lookupUrl, ClientConfiguration().setOperationTimeoutSeconds(2));
+    EXPECT_CALL(*clientImpl, getConnection).Times(AtLeast(2 * 2));
+    ON_CALL(*clientImpl, getConnection).WillByDefault([](const std::string &topic, size_t index) {
+        return GetConnectionFuture::failed(ResultRetryable);
+    });
+
+    auto topic = "client-test-retry-timeout";
+    {
+        MockClientImpl::SyncOpResult result = clientImpl->createProducer(topic);
+        ASSERT_EQ(ResultTimeout, result.result);
+        ASSERT_TRUE(result.timeMs >= 2000 && result.timeMs < 2100) << "producer: " << result.timeMs << " ms";
+    }
+    {
+        MockClientImpl::SyncOpResult result = clientImpl->subscribe(topic);
+        ASSERT_EQ(ResultTimeout, result.result);
+        ASSERT_TRUE(result.timeMs >= 2000 && result.timeMs < 2100) << "consumer: " << result.timeMs << " ms";
+    }
+
+    ASSERT_EQ(ResultOk, clientImpl->close());
+}
+
+TEST(ClientTest, testNoRetry) {
+    auto clientImpl =
+        std::make_shared<MockClientImpl>(lookupUrl, ClientConfiguration().setOperationTimeoutSeconds(100));
+    EXPECT_CALL(*clientImpl, getConnection).Times(2);
+    ON_CALL(*clientImpl, getConnection).WillByDefault([](const std::string &, size_t) {
+        return GetConnectionFuture::failed(ResultAuthenticationError);
+    });
+
+    auto topic = "client-test-no-retry";
+    {
+        MockClientImpl::SyncOpResult result = clientImpl->createProducer(topic);
+        ASSERT_EQ(ResultAuthenticationError, result.result);
+        ASSERT_TRUE(result.timeMs < 1000) << "producer: " << result.timeMs << " ms";
+    }
+    {
+        MockClientImpl::SyncOpResult result = clientImpl->subscribe(topic);
+        LOG_INFO("It takes " << result.timeMs << " ms to subscribe");
+        ASSERT_EQ(ResultAuthenticationError, result.result);
+        ASSERT_TRUE(result.timeMs < 1000) << "consumer: " << result.timeMs << " ms";
+    }
+}
diff --git a/tests/MockClientImpl.h b/tests/MockClientImpl.h

Original file line number	Diff line number	Diff line change
`@@ -278,7 +278,7 @@ ClientConnection::~ClientConnection() {`
`278`	`278`	`void ClientConnection::handlePulsarConnected(const proto::CommandConnected& cmdConnected) {`
`279`	`279`	`if (!cmdConnected.has_server_version()) {`
`280`	`280`	`LOG_ERROR(cnxString_ << "Server version is not set");`
`281`		`- close();`
	`281`	`+ close(ResultUnknownError);`
`282`	`282`	`return;`
`283`	`283`	`}`
`284`	`284`
`@@ -451,7 +451,7 @@ void ClientConnection::handleTcpConnected(const ASIO_ERROR& err, tcp::resolver::`
`451`	`451`	`Url service_url;`
`452`	`452`	`if (!Url::parse(physicalAddress_, service_url)) {`
`453`	`453`	`LOG_ERROR(cnxString_ << "Invalid Url, unable to parse: " << err << " " << err.message());`
`454`		`- close();`
	`454`	`+ close(ResultInvalidUrl);`
`455`	`455`	`return;`
`456`	`456`	`}`
`457`	`457`	`}`
`@@ -489,12 +489,7 @@ void ClientConnection::handleTcpConnected(const ASIO_ERROR& err, tcp::resolver::`
`489`	`489`	`}`
`490`	`490`	`});`
`491`	`491`	`} else {`
`492`		`- if (err == ASIO::error::operation_aborted) {`
`493`		`- // TCP connect timeout, which is not retryable`
`494`		`- close();`
`495`		`- } else {`
`496`		`- close(ResultRetryable);`
`497`		`- }`
	`492`	`+ close(ResultRetryable);`
`498`	`493`	`}`
`499`	`494`	`} else {`
`500`	`495`	`LOG_ERROR(cnxString_ << "Failed to establish connection: " << err.message());`
`@@ -504,8 +499,8 @@ void ClientConnection::handleTcpConnected(const ASIO_ERROR& err, tcp::resolver::`
`504`	`499`
`505`	`500`	`void ClientConnection::handleHandshake(const ASIO_ERROR& err) {`
`506`	`501`	`if (err) {`
`507`		`- LOG_ERROR(cnxString_ << "Handshake failed: " << err.message());`
`508`		`- close();`
	`502`	`+ LOG_WARN(cnxString_ << "Handshake failed: " << err.message());`
	`503`	`+ close(ResultRetryable);`
`509`	`504`	`return;`
`510`	`505`	`}`
`511`	`506`
`@@ -532,7 +527,7 @@ void ClientConnection::handleSentPulsarConnect(const ASIO_ERROR& err, const Shar`
`532`	`527`	`}`
`533`	`528`	`if (err) {`
`534`	`529`	`LOG_ERROR(cnxString_ << "Failed to establish connection: " << err.message());`
`535`		`- close();`
	`530`	`+ close(ResultRetryable);`
`536`	`531`	`return;`
`537`	`532`	`}`
`538`	`533`
`@@ -546,7 +541,7 @@ void ClientConnection::handleSentAuthResponse(const ASIO_ERROR& err, const Share`
`546`	`541`	`}`
`547`	`542`	`if (err) {`
`548`	`543`	`LOG_WARN(cnxString_ << "Failed to send auth response: " << err.message());`
`549`		`- close();`
	`544`	`+ close(ResultRetryable);`
`550`	`545`	`return;`
`551`	`546`	`}`
`552`	`547`	`}`
`@@ -567,14 +562,14 @@ void ClientConnection::tcpConnectAsync() {`
`567`	`562`	`std::string hostUrl = isSniProxy_ ? proxyServiceUrl_ : physicalAddress_;`
`568`	`563`	`if (!Url::parse(hostUrl, service_url)) {`
`569`	`564`	`LOG_ERROR(cnxString_ << "Invalid Url, unable to parse: " << err << " " << err.message());`
`570`		`- close();`
	`565`	`+ close(ResultInvalidUrl);`
`571`	`566`	`return;`
`572`	`567`	`}`
`573`	`568`
`574`	`569`	`if (service_url.protocol() != "pulsar" && service_url.protocol() != "pulsar+ssl") {`
`575`	`570`	`LOG_ERROR(cnxString_ << "Invalid Url protocol '" << service_url.protocol()`
`576`	`571`	`<< "'. Valid values are 'pulsar' and 'pulsar+ssl'");`
`577`		`- close();`
	`572`	`+ close(ResultInvalidUrl);`
`578`	`573`	`return;`
`579`	`574`	`}`
`580`	`575`
`@@ -593,7 +588,7 @@ void ClientConnection::handleResolve(const ASIO_ERROR& err, tcp::resolver::itera`
`593`	`588`	`if (err) {`
`594`	`589`	`std::string hostUrl = isSniProxy_ ? cnxString_ : proxyServiceUrl_;`
`595`	`590`	`LOG_ERROR(hostUrl << "Resolve error: " << err << " : " << err.message());`
`596`		`- close();`
	`591`	`+ close(ResultConnectError);`
`597`	`592`	`return;`
`598`	`593`	`}`
`599`	`594`
`@@ -629,8 +624,8 @@ void ClientConnection::handleResolve(const ASIO_ERROR& err, tcp::resolver::itera`
`629`	`624`	`}`
`630`	`625`	`});`
`631`	`626`	`} else {`
`632`		`- LOG_WARN(cnxString_ << "No IP address found");`
`633`		`- close();`
	`627`	`+ LOG_ERROR(cnxString_ << "No IP address found");`
	`628`	`+ close(ResultConnectError);`
`634`	`629`	`return;`
`635`	`630`	`}`
`636`	`631`	`}`
`@@ -885,7 +880,7 @@ void ClientConnection::handleIncomingCommand(BaseCommand& incomingCmd) {`
`885`	`880`	`// Handle Pulsar Connected`
`886`	`881`	`if (incomingCmd.type() != BaseCommand::CONNECTED) {`
`887`	`882`	`// Wrong cmd`
`888`		`- close();`
	`883`	`+ close(ResultRetryable);`
`889`	`884`	`} else {`
`890`	`885`	`handlePulsarConnected(incomingCmd.connected());`
`891`	`886`	`}`
Original file line number	Diff line number	Diff line change
`@@ -157,7 +157,7 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien`
`157`	`157`	`*`
`158`	`158`	* `detach` should only be false when the connection pool is closed.
`159`	`159`	`*/`
`160`		`- void close(Result result = ResultConnectError, bool detach = true);`
	`160`	`+ void close(Result result, bool detach = true);`
`161`	`161`
`162`	`162`	`bool isClosed() const;`
`163`	`163`
Original file line number	Diff line number	Diff line change
`@@ -517,7 +517,7 @@ void ClientImpl::handleConsumerCreated(Result result, ConsumerImplBaseWeakPtr co`
`517`	`517`	`}`
`518`	`518`	`}`
`519`	`519`
`520`		`-Future<Result, ClientConnectionPtr> ClientImpl::getConnection(const std::string& topic, size_t key) {`
	`520`	`+GetConnectionFuture ClientImpl::getConnection(const std::string& topic, size_t key) {`
`521`	`521`	`Promise<Result, ClientConnectionPtr> promise;`
`522`	`522`
`523`	`523`	`const auto topicNamePtr = TopicName::get(topic);`
`@@ -562,7 +562,7 @@ const std::string& ClientImpl::getPhysicalAddress(const std::string& logicalAddr`
`562`	`562`	`}`
`563`	`563`	`}`
`564`	`564`
`565`		`-Future<Result, ClientConnectionPtr> ClientImpl::connect(const std::string& logicalAddress, size_t key) {`
	`565`	`+GetConnectionFuture ClientImpl::connect(const std::string& logicalAddress, size_t key) {`
`566`	`566`	`const auto& physicalAddress = getPhysicalAddress(logicalAddress);`
`567`	`567`	`Promise<Result, ClientConnectionPtr> promise;`
`568`	`568`	`pool_.getConnectionAsync(logicalAddress, physicalAddress, key)`
Original file line number	Diff line number	Diff line change
`@@ -277,7 +277,7 @@ void ConsumerImpl::connectionFailed(Result result) {`
`277`	`277`	`// Keep a reference to ensure object is kept alive`
`278`	`278`	`auto ptr = get_shared_this_ptr();`
`279`	`279`
`280`		`- if (consumerCreatedPromise_.setFailed(result)) {`
	`280`	`+ if (!isResultRetryable(result) && consumerCreatedPromise_.setFailed(result)) {`
`281`	`281`	`state_ = Failed;`
`282`	`282`	`}`
`283`	`283`	`}`
Original file line number	Diff line number	Diff line change
`@@ -176,7 +176,7 @@ void ProducerImpl::connectionFailed(Result result) {`
`176`	`176`	`// if producers are lazy, then they should always try to restart`
`177`	`177`	`// so don't change the state and allow reconnections`
`178`	`178`	`return;`
`179`		`- } else if (producerCreatedPromise_.setFailed(result)) {`
	`179`	`+ } else if (!isResultRetryable(result) && producerCreatedPromise_.setFailed(result)) {`
`180`	`180`	`state_ = Failed;`
`181`	`181`	`}`
`182`	`182`	`}`