Fix creating producer or consumer is not retried for connection failure

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`.

Author: BewareMyPower

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)

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();

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;
     }
 }

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

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;
     }

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_;

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;
     }
 }

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/";
@@ -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";
+    }
+}

tests/MockClientImpl.h

@@ -0,0 +1,75 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+#pragma once
+
+#include <gmock/gmock.h>
+
+#include <chrono>
+#include <future>
+
+#include "lib/ClientImpl.h"
+
+namespace pulsar {
+
+class MockClientImpl : public ClientImpl {
+   public:
+    struct SyncOpResult {
+        Result result;
+        long timeMs;
+    };
+    MockClientImpl(const std::string& serviceUrl, ClientConfiguration conf = {})
+        : ClientImpl(serviceUrl, conf) {}
+
+    MOCK_METHOD((Future<Result, ClientConnectionPtr>), getConnection, (const std::string&, size_t),
+                (override));
+
+    SyncOpResult createProducer(const std::string& topic) {
+        using namespace std::chrono;
+        auto start = high_resolution_clock::now();
+        std::promise<SyncOpResult> promise;
+        createProducerAsync(topic, {}, [&start, &promise](Result result, Producer) {
+            auto timeMs = duration_cast<milliseconds>(high_resolution_clock::now() - start).count();
+            promise.set_value({result, timeMs});
+        });
+        return promise.get_future().get();
+    }
+
+    SyncOpResult subscribe(const std::string& topic) {
+        using namespace std::chrono;
+        auto start = std::chrono::high_resolution_clock::now();
+        std::promise<SyncOpResult> promise;
+        subscribeAsync(topic, "sub", {}, [&start, &promise](Result result, Consumer) {
+            auto timeMs = duration_cast<milliseconds>(high_resolution_clock::now() - start).count();
+            promise.set_value({result, timeMs});
+        });
+        return promise.get_future().get();
+    }
+
+    GetConnectionFuture getConnectionReal(const std::string& topic, size_t key) {
+        return ClientImpl::getConnection(topic, key);
+    }
+
+    Result close() {
+        std::promise<Result> promise;
+        closeAsync([&promise](Result result) { promise.set_value(result); });
+        return promise.get_future().get();
+    }
+};
+
+}  // namespace pulsar