feat: add an interface for rpc_holder to set error

include/dsn/cpp/rpc_holder.h

@@ -114,6 +114,12 @@ class rpc_holder
         return _i->thrift_response;
     }
 
+    dsn::error_code &error() const
+    {
+        dassert(_i, "rpc_holder is uninitialized");
+        return _i->rpc_error;
+    }
+
     message_ex *dsn_request() const
     {
         dassert(_i, "rpc_holder is uninitialized");
@@ -291,7 +297,7 @@ class rpc_holder
 
             message_ex *dsn_response = dsn_request->create_response();
             marshall(dsn_response, thrift_response);
-            dsn_rpc_reply(dsn_response);
+            dsn_rpc_reply(dsn_response, rpc_error);
         }
 
         ~internal()
@@ -305,6 +311,7 @@ class rpc_holder
         message_ex *dsn_request;
         std::unique_ptr<TRequest> thrift_request;
         TResponse thrift_response;
+        dsn::error_code rpc_error = dsn::ERR_OK;
 
         bool auto_reply;
     };

src/runtime/test/rpc_holder_test.cpp

@@ -70,6 +70,19 @@ TEST(rpc_holder, construct)
         ASSERT_TRUE(rpc.is_initialized());
         ASSERT_EQ(rpc.request().app_name, "test");
     }
+
+    {
+        auto request = make_unique<configuration_query_by_index_request>();
+        t_rpc rpc(std::move(request), RPC_CM_QUERY_PARTITION_CONFIG_BY_INDEX);
+        ASSERT_EQ(rpc.error(), ERR_OK);
+        ASSERT_TRUE(rpc.is_initialized());
+
+        rpc.error() = ERR_BUSY;
+        ASSERT_EQ(rpc.error(), ERR_BUSY);
+
+        rpc.error() = ERR_ADDRESS_ALREADY_USED;
+        ASSERT_EQ(rpc.error(), ERR_ADDRESS_ALREADY_USED);
+    }
 }
 
 TEST(rpc_holder, mock_rpc_call)
@@ -87,6 +100,25 @@ TEST(rpc_holder, mock_rpc_call)
         ASSERT_EQ(mail_box.size(), 10);
     }
 
+    // test in error cases
+    RPC_MOCKING(t_rpc)
+    {
+        auto &mail_box = t_rpc::mail_box();
+
+        for (int i = 0; i < 10; i++) {
+            auto request = make_unique<configuration_query_by_index_request>();
+            t_rpc rpc(std::move(request), RPC_CM_QUERY_PARTITION_CONFIG_BY_INDEX);
+            rpc.error() = ERR_BUSY;
+            rpc.call(rpc_address("127.0.0.1", 12321), nullptr, [](error_code) {});
+        }
+
+        ASSERT_EQ(mail_box.size(), 10);
+
+        for (const auto &iter : mail_box) {
+            ASSERT_EQ(iter.error(), ERR_BUSY);
+        }
+    }
+
     // instances of rpc mocking are independent
     RPC_MOCKING(t_rpc)
     {