Fix retriable errors not handled well when creating producer or consumer (#293)

* Fix retriable errors not handled well when subscribing

Fixes #292

### Motivation

When a consumer failed to subscribe due to a retriable error, the time
point comparation is wrong:

https://github.com/apache/pulsar-client-cpp/blob/633f4bbe8c182128da09803172676b9d6af05057/lib/ConsumerImpl.cc#L321

`creationTimestamp_ + operationTimeut_` is the deadline,
`TimeUtils::now()` is the current time, we should use `>` instead of `<`
here to compare them. Otherwise, if the consumer encountered a retriable
error and the deadline is not exceeded, the consumer won't reconnect and
fail with `ResultRetryable`.

### Modifications

Reverse the comparation between the deadline and the current time. When
it times out, completing the future with `ResultTimeout` instead of the
`result` itself, which is always `ResultRetryable`.

Add `ConsumerTest.testRetrySubscribe` to verify this change.

### TODO

Support configuring the operation timeout in milliseconds.

* Fix the same error for producer as well

* Add result conversion

* Reduce duplicated code

Author: BewareMyPower

lib/ConsumerImpl.cc

@@ -318,8 +318,9 @@ void ConsumerImpl::handleCreateConsumer(const ClientConnectionPtr& cnx, Result r
             scheduleReconnection(get_shared_this_ptr());
         } else {
             // Consumer was not yet created, retry to connect to broker if it's possible
-            if (isRetriableError(result) && (creationTimestamp_ + operationTimeut_ < TimeUtils::now())) {
-                LOG_WARN(getName() << "Temporary error in creating consumer : " << strResult(result));
+            result = convertToTimeoutIfNecessary(result, creationTimestamp_);
+            if (result == ResultRetryable) {
+                LOG_WARN(getName() << "Temporary error in creating consumer: " << strResult(result));
                 scheduleReconnection(get_shared_this_ptr());
             } else {
                 LOG_ERROR(getName() << "Failed to create consumer: " << strResult(result));

lib/HandlerBase.cc

@@ -138,8 +138,6 @@ void HandlerBase::handleDisconnection(Result result, ClientConnectionWeakPtr con
     }
 }
 
-bool HandlerBase::isRetriableError(Result result) { return result == ResultRetryable; }
-
 void HandlerBase::scheduleReconnection(HandlerBasePtr handler) {
     const auto state = handler->state_.load();
     if (state == Pending || state == Ready) {
@@ -164,4 +162,12 @@ void HandlerBase::handleTimeout(const boost::system::error_code& ec, HandlerBase
     }
 }
 
+Result HandlerBase::convertToTimeoutIfNecessary(Result result, ptime startTimestamp) const {
+    if (result == ResultRetryable && (TimeUtils::now() - startTimestamp >= operationTimeut_)) {
+        return ResultTimeout;
+    } else {
+        return result;
+    }
+}
+
 }  // namespace pulsar

lib/HandlerBase.h

@@ -69,11 +69,6 @@ class HandlerBase {
      */
     static void scheduleReconnection(HandlerBasePtr handler);
 
-    /*
-     * Should we retry in error that are transient
-     */
-    bool isRetriableError(Result result);
-
     /**
      * Do some cleanup work before changing `connection_` to `cnx`.
      *
@@ -127,6 +122,8 @@ class HandlerBase {
     Backoff backoff_;
     uint64_t epoch_;
 
+    Result convertToTimeoutIfNecessary(Result result, ptime startTimestamp) const;
+
    private:
     DeadlineTimerPtr timer_;
 

lib/ProducerImpl.cc

@@ -281,7 +281,8 @@ void ProducerImpl::handleCreateProducer(const ClientConnectionPtr& cnx, Result r
             scheduleReconnection(shared_from_this());
         } else {
             // Producer was not yet created, retry to connect to broker if it's possible
-            if (isRetriableError(result) && (creationTimestamp_ + operationTimeut_ < TimeUtils::now())) {
+            result = convertToTimeoutIfNecessary(result, creationTimestamp_);
+            if (result == ResultRetryable) {
                 LOG_WARN(getName() << "Temporary error in creating producer: " << strResult(result));
                 scheduleReconnection(shared_from_this());
             } else {

tests/ConsumerTest.cc

@@ -1321,4 +1321,18 @@ TEST(ConsumerTest, testNotSetSubscriptionName) {
     client.close();
 }
 
+TEST(ConsumerTest, testRetrySubscribe) {
+    Client client{lookupUrl};
+    for (int i = 0; i < 10; i++) {
+        // "Subscription is fenced" error might happen here because the previous seek operation might not be
+        // done in broker, the consumer should retry until timeout
+        Consumer consumer;
+        ASSERT_EQ(client.subscribe("test-close-before-seek-done", "sub", consumer), ResultOk);
+        consumer.seekAsync(MessageId::earliest(), [](Result) {});
+        consumer.close();
+    }
+    // TODO: Currently it's hard to test the timeout error without configuring the operation timeout in
+    // milliseconds
+}
+
 }  // namespace pulsar