fix(ddl_client): sleep for a while before retry once meta server is busy

src/client/replication_ddl_client.cpp

@@ -52,11 +52,23 @@
 #include "runtime/api_layer1.h"
 #include "runtime/rpc/group_address.h"
 #include "utils/error_code.h"
+#include "utils/flags.h"
 #include "utils/fmt_logging.h"
 #include "utils/output_utils.h"
 #include "utils/time_utils.h"
 #include "utils/utils.h"
 
+DSN_DEFINE_uint32(ddl_client,
+                  ddl_client_max_attempt_count,
+                  3,
+                  "The max count that attempt for failed requests to meta server.");
+
+DSN_DEFINE_uint32(ddl_client,
+                  ddl_client_busy_retry_interval_ms,
+                  10 * 1000,
+                  "The retry interval after receiving ERR_BUSY_CREATING or ERR_BUSY_DROPPING from"
+                  "meta server.");
+
 namespace dsn {
 namespace replication {
 
@@ -1434,23 +1446,83 @@ bool replication_ddl_client::valid_app_char(int c)
     return (bool)std::isalnum(c) || c == '_' || c == '.' || c == ':';
 }
 
+namespace {
+
+bool is_busy(const dsn::error_code &err)
+{
+    return err == dsn::ERR_BUSY_CREATING || err == dsn::ERR_BUSY_DROPPING;
+}
+
+bool need_retry(uint32_t attempt_count, uint32_t busy_attempt_count, const dsn::error_code &err)
+{
+    if (err == dsn::ERR_OK) {
+        return false;
+    }
+
+    if (attempt_count < FLAGS_ddl_client_max_attempt_count) {
+        return true;
+    }
+
+    if (!is_busy(err)) {
+        return false;
+    }
+
+    return busy_attempt_count < FLAGS_ddl_client_max_attempt_count;
+}
+
+bool should_sleep_before_retry(const dsn::error_code &err, uint32_t &sleep_ms)
+{
+    if (is_busy(err)) {
+        sleep_ms = FLAGS_ddl_client_busy_retry_interval_ms;
+        return true;
+    }
+
+    return false;
+}
+
+} // anonymous namespace
+
 void replication_ddl_client::end_meta_request(const rpc_response_task_ptr &callback,
-                                              int retry_times,
-                                              error_code err,
+                                              uint32_t attempt_count,
+                                              uint32_t busy_attempt_count,
+                                              const error_code &err,
                                               dsn::message_ex *request,
                                               dsn::message_ex *resp)
 {
-    if (err != dsn::ERR_OK && retry_times < 2) {
-        rpc::call(_meta_server,
-                  request,
-                  &_tracker,
-                  [this, retry_times, callback](
-                      error_code err, dsn::message_ex *request, dsn::message_ex *response) mutable {
-                      end_meta_request(callback, retry_times + 1, err, request, response);
-                  });
-    } else {
-        callback->enqueue(err, (message_ex *)resp);
+    ++attempt_count;
+    if (is_busy(err)) {
+        ++busy_attempt_count;
     }
+
+    LOG_INFO("attempt {}: attempt_count={}, busy_attempt_count={}, max_replica_count={}, err={}",
+             request->local_rpc_code,
+             attempt_count,
+             busy_attempt_count,
+             FLAGS_ddl_client_max_attempt_count,
+             err);
+
+    if (!need_retry(attempt_count, busy_attempt_count, err)) {
+        callback->enqueue(err, (message_ex *)resp);
+        return;
+    }
+
+    uint32_t sleep_ms = 0;
+    if (should_sleep_before_retry(err, sleep_ms)) {
+        LOG_WARNING("sleep {} milliseconds before launch another attempt for {}: err={}",
+                    sleep_ms,
+                    request->local_rpc_code,
+                    err);
+        std::this_thread::sleep_for(std::chrono::milliseconds(sleep_ms));
+    }
+
+    rpc::call(_meta_server,
+              request,
+              &_tracker,
+              [this, attempt_count, busy_attempt_count, callback](
+                  error_code err, dsn::message_ex *request, dsn::message_ex *response) mutable {
+                  end_meta_request(
+                      callback, attempt_count, busy_attempt_count, err, request, response);
+              });
 }
 
 dsn::error_code replication_ddl_client::get_app_envs(const std::string &app_name,

src/client/replication_ddl_client.h

@@ -257,8 +257,9 @@ class replication_ddl_client
     bool static valid_app_char(int c);
 
     void end_meta_request(const rpc_response_task_ptr &callback,
-                          int retry_times,
-                          error_code err,
+                          uint32_t attempt_count,
+                          uint32_t busy_attempt_count,
+                          const error_code &err,
                           dsn::message_ex *request,
                           dsn::message_ex *resp);
 
@@ -278,7 +279,7 @@ class replication_ddl_client
                   &_tracker,
                   [this, task](
                       error_code err, dsn::message_ex *request, dsn::message_ex *response) mutable {
-                      end_meta_request(std::move(task), 0, err, request, response);
+                      end_meta_request(std::move(task), 0, 0, err, request, response);
                   });
         return task;
     }

src/runtime/task/task_code.h

@@ -200,4 +200,5 @@ DEFINE_TASK_CODE(TASK_CODE_INVALID, TASK_PRIORITY_COMMON, THREAD_POOL_DEFAULT)
 // define a task_code "task_code_inlined", it's mainly used in situations when you want execute
 // a task with "inline" mode.
 DEFINE_TASK_CODE(TASK_CODE_EXEC_INLINED, TASK_PRIORITY_COMMON, THREAD_POOL_DEFAULT)
-}
+
+} // namespace dsn