Fix creating producer or consumer is not retried for connection failure

BewareMyPower · BewareMyPower · commit 8528a2328626 · 2024-02-06T21:09:00.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`.
diff --git a/lib/ClientConnection.cc b/lib/ClientConnection.cc
@@ -81,9 +81,9 @@ static Result getResult(ServerError serverError, const std::string& message) {
             return ResultConsumerBusy;
 
         case ServiceNotReady:
-            // If the error is not caused by a PulsarServerException, treat it as retryable.
-            return (message.find("PulsarServerException") == std::string::npos) ? ResultRetryable
-                                                                                : ResultServiceUnitNotReady;
+            return (message.find("the broker do not have test listener") != std::string::npos)
+                       ? ResultConnectError
+                       : ResultServiceUnitNotReady;
 
         case ProducerBlockedQuotaExceededError:
             return ResultProducerBlockedQuotaExceededError;
diff --git a/lib/ClientConnectionAdaptor.h b/lib/ClientConnectionAdaptor.h
@@ -37,12 +37,14 @@ inline void checkServerError(Connection& connection, ServerError error, const st
             //   "Namespace bundle ... is being unloaded"
             //   "KeeperException$..."
             //   "Failed to acquire ownership for namespace bundle ..."
+            //   "the broker do not have test listener"
             // Before https://github.com/apache/pulsar/pull/21211, the error of the 1st and 2nd messages
             // is ServiceNotReady. Before https://github.com/apache/pulsar/pull/21993, the error of the 3rd
             // message is ServiceNotReady.
             if (message.find("Failed to acquire ownership") == std::string::npos &&
                 message.find("KeeperException") == std::string::npos &&
-                message.find("is being unloaded") == std::string::npos) {
+                message.find("is being unloaded") == std::string::npos &&
+                message.find("the broker do not have test listener") == std::string::npos) {
                 connection.close(ResultDisconnected);
             }
             break;
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;
     }
@@ -108,7 +123,7 @@ void HandlerBase::grabCnx(const boost::optional<std::string>& assignedBrokerUrl)
             connectionOpened(cnx).addListener([this, self](Result result, bool) {
                 // Do not use bool, only Result.
                 reconnectionPending_ = false;
-                if (isResultRetryable(result)) {
+                if (result != ResultOk && isResultRetryable(result)) {
                     scheduleReconnection();
                 }
             });
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/lib/ResultUtils.h b/lib/ResultUtils.h
@@ -18,12 +18,39 @@
  */
 #pragma once
 
+#include <assert.h>
 #include <pulsar/Result.h>
 
+#include <unordered_set>
+
 namespace pulsar {
 
 inline bool isResultRetryable(Result result) {
-    return result == ResultRetryable || result == ResultDisconnected;
+    assert(result != ResultOk);
+    if (result == ResultRetryable || result == ResultDisconnected) {
+        return true;
+    }
+
+    static const std::unordered_set<Result> fatalResults{ResultConnectError,
+                                                         ResultTimeout,
+                                                         ResultAuthenticationError,
+                                                         ResultAuthorizationError,
+                                                         ResultInvalidUrl,
+                                                         ResultInvalidConfiguration,
+                                                         ResultIncompatibleSchema,
+                                                         ResultTopicNotFound,
+                                                         ResultOperationNotSupported,
+                                                         ResultNotAllowedError,
+                                                         ResultChecksumError,
+                                                         ResultCryptoError,
+                                                         ResultConsumerAssignError,
+                                                         ResultProducerBusy,
+                                                         ResultConsumerBusy,
+                                                         ResultLookupError,
+                                                         ResultTooManyLookupRequestException,
+                                                         ResultProducerBlockedQuotaExceededException,
+                                                         ResultProducerBlockedQuotaExceededError};
+    return fatalResults.find(result) == fatalResults.cend();
 }
 
 }  // namespace pulsar
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/";
@@ -248,7 +250,7 @@ TEST(ClientTest, testWrongListener) {
 
     Client client(lookupUrl, ClientConfiguration().setListenerName("test"));
     Producer producer;
-    ASSERT_EQ(ResultServiceUnitNotReady, client.createProducer(topic, producer));
+    ASSERT_EQ(ResultConnectError, client.createProducer(topic, producer));
     ASSERT_EQ(ResultProducerNotInitialized, producer.close());
     ASSERT_EQ(PulsarFriend::getProducers(client).size(), 0);
     ASSERT_EQ(ResultOk, client.close());
@@ -257,7 +259,7 @@ TEST(ClientTest, testWrongListener) {
     // creation of Consumer or Reader could fail with ResultConnectError.
     client = Client(lookupUrl, ClientConfiguration().setListenerName("test"));
     Consumer consumer;
-    ASSERT_EQ(ResultServiceUnitNotReady, client.subscribe(topic, "sub", consumer));
+    ASSERT_EQ(ResultConnectError, client.subscribe(topic, "sub", consumer));
     ASSERT_EQ(ResultConsumerNotInitialized, consumer.close());
 
     ASSERT_EQ(PulsarFriend::getConsumers(client).size(), 0);
@@ -266,7 +268,7 @@ TEST(ClientTest, testWrongListener) {
     client = Client(lookupUrl, ClientConfiguration().setListenerName("test"));
 
     Consumer multiTopicsConsumer;
-    ASSERT_EQ(ResultServiceUnitNotReady,
+    ASSERT_EQ(ResultConnectError,
               client.subscribe({topic + "-partition-0", topic + "-partition-1", topic + "-partition-2"},
                                "sub", multiTopicsConsumer));
 
@@ -278,7 +280,7 @@ TEST(ClientTest, testWrongListener) {
 
     // Currently Reader can only read a non-partitioned topic in C++ client
     Reader reader;
-    ASSERT_EQ(ResultServiceUnitNotReady,
+    ASSERT_EQ(ResultConnectError,
               client.createReader(topic + "-partition-0", MessageId::earliest(), {}, reader));
     ASSERT_EQ(ResultConsumerNotInitialized, reader.close());
     ASSERT_EQ(PulsarFriend::getConsumers(client).size(), 0);
@@ -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
diff --git a/tests/extensibleLM/ExtensibleLoadManagerTest.cc b/tests/extensibleLM/ExtensibleLoadManagerTest.cc

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