This is an automated email from the ASF dual-hosted git repository.
wangdan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pegasus.git
The following commit(s) were added to refs/heads/master by this push:
new 633042764 fix(duplication): dup would get stuck in `DS_PREPARE`
indefinitely once some error occurred during creating follower table (#2144)
633042764 is described below
commit 6330427640cb7277cbce3b082f50867317b460d3
Author: Dan Wang <wang...@apache.org>
AuthorDate: Fri Nov 15 12:10:43 2024 +0800
fix(duplication): dup would get stuck in `DS_PREPARE` indefinitely once
some error occurred during creating follower table (#2144)
Fix https://github.com/apache/incubator-pegasus/issues/2146.
As is described in issue, once there are some errors occurred during the
follower
table is being created, the follower cluster might replied with
`ERR_APP_EXIST`
to the source cluster. Then, the source cluster would continue to send the
same
request to create table on the follower cluster again. As such, the dup
would get
stuck in the current status, namely `DS_PREPARE`, and could not forward to
the
next status.
The core idea to solve this problem is that we should make the request that
create
table on follower cluster idempotent. The follower cluster should process
the repeated
requests correctly. To implement this, an environment variable marking
creating status
is added and sent together with the request to the follower cluster. The
follower cluster
would choose to commit it to meta data on both remote storage and local
memory, or
just ignore it, or reply to source cluster with some error code.
---
.clang-tidy | 2 +-
build_tools/clang_tidy.py | 1 +
src/common/duplication_common.cpp | 13 +-
src/common/duplication_common.h | 10 +-
src/meta/app_env_validator.cpp | 10 +-
src/meta/duplication/meta_duplication_service.cpp | 218 +++++++++---
src/meta/duplication/meta_duplication_service.h | 33 +-
src/meta/server_state.cpp | 387 ++++++++++++++++-----
src/meta/server_state.h | 20 +-
src/meta/test/meta_app_operation_test.cpp | 307 ++++++++++++----
src/meta/test/meta_duplication_service_test.cpp | 145 ++++----
src/meta/test/meta_state/meta_state_service.cpp | 7 +-
src/replica/duplication/replica_follower.cpp | 6 +-
.../duplication/test/replica_follower_test.cpp | 35 +-
src/replica/replica_stub.cpp | 6 +-
src/rpc/rpc_stream.h | 30 +-
src/utils/binary_writer.cpp | 36 +-
src/utils/binary_writer.h | 13 +-
src/utils/fmt_utils.h | 2 +-
src/utils/ports.h | 18 +-
20 files changed, 938 insertions(+), 361 deletions(-)
diff --git a/.clang-tidy b/.clang-tidy
index 888e91b95..f6da75564 100644
--- a/.clang-tidy
+++ b/.clang-tidy
@@ -20,7 +20,7 @@
CheckOptions: []
# Disable some checks that are not useful for us now.
# They are sorted by names, and should be consistent to
build_tools/clang_tidy.py.
-Checks:
'abseil-*,boost-*,bugprone-*,cert-*,clang-analyzer-*,concurrency-*,cppcoreguidelines-*,darwin-*,fuchsia-*,google-*,hicpp-*,linuxkernel-*,llvm-*,misc-*,modernize-*,performance-*,portability-*,readability-*,-bugprone-easily-swappable-parameters,-bugprone-lambda-function-name,-bugprone-macro-parentheses,-cert-err58-cpp,-concurrency-mt-unsafe,-cppcoreguidelines-avoid-c-arrays,-cppcoreguidelines-avoid-magic-numbers,-cppcoreguidelines-avoid-non-const-global-variables,-cppcoreguidelines
[...]
+Checks:
'abseil-*,boost-*,bugprone-*,cert-*,clang-analyzer-*,concurrency-*,cppcoreguidelines-*,darwin-*,fuchsia-*,google-*,hicpp-*,linuxkernel-*,llvm-*,misc-*,modernize-*,performance-*,portability-*,readability-*,-bugprone-easily-swappable-parameters,-bugprone-lambda-function-name,-bugprone-macro-parentheses,-cert-err58-cpp,-concurrency-mt-unsafe,-cppcoreguidelines-avoid-c-arrays,-cppcoreguidelines-avoid-magic-numbers,-cppcoreguidelines-avoid-non-const-global-variables,-cppcoreguidelines
[...]
ExtraArgs:
ExtraArgsBefore: []
FormatStyle: none
diff --git a/build_tools/clang_tidy.py b/build_tools/clang_tidy.py
index b2ef72d9b..e5ef014b8 100755
--- a/build_tools/clang_tidy.py
+++ b/build_tools/clang_tidy.py
@@ -90,6 +90,7 @@ def run_tidy(sha="HEAD", is_rev_range=False):
"-modernize-avoid-c-arrays,"
"-modernize-replace-disallow-copy-and-assign-macro,"
"-modernize-use-trailing-return-type,"
+ "-performance-unnecessary-value-param,"
"-readability-function-cognitive-complexity,"
"-readability-identifier-length,"
"-readability-magic-numbers,"
diff --git a/src/common/duplication_common.cpp
b/src/common/duplication_common.cpp
index 4dc5323c6..cc49165f9 100644
--- a/src/common/duplication_common.cpp
+++ b/src/common/duplication_common.cpp
@@ -55,12 +55,17 @@ namespace replication {
const std::string duplication_constants::kDuplicationCheckpointRootDir
/*NOLINT*/ = "duplication";
const std::string duplication_constants::kClustersSectionName /*NOLINT*/ =
"pegasus.clusters";
-const std::string duplication_constants::kDuplicationEnvMasterClusterKey
/*NOLINT*/ =
+const std::string duplication_constants::kEnvMasterClusterKey /*NOLINT*/ =
"duplication.master_cluster";
-const std::string duplication_constants::kDuplicationEnvMasterMetasKey
/*NOLINT*/ =
- "duplication.master_metas";
-const std::string duplication_constants::kDuplicationEnvMasterAppNameKey
/*NOLINT*/ =
+const std::string duplication_constants::kEnvMasterMetasKey /*NOLINT*/ =
"duplication.master_metas";
+const std::string duplication_constants::kEnvMasterAppNameKey /*NOLINT*/ =
"duplication.master_app_name";
+const std::string duplication_constants::kEnvFollowerAppStatusKey /*NOLINT*/
+ = "duplication.follower_app_status";
+const std::string duplication_constants::kEnvFollowerAppStatusCreating
/*NOLINT*/
+ = "creating";
+const std::string duplication_constants::kEnvFollowerAppStatusCreated
/*NOLINT*/
+ = "created";
/*extern*/ const char *duplication_status_to_string(duplication_status::type
status)
{
diff --git a/src/common/duplication_common.h b/src/common/duplication_common.h
index 861335733..0b42daef4 100644
--- a/src/common/duplication_common.h
+++ b/src/common/duplication_common.h
@@ -81,12 +81,16 @@ struct duplication_constants
const static std::string kDuplicationCheckpointRootDir;
const static std::string kClustersSectionName;
// These will fill into app env and mark one app as a "follower app" and
record master info
- const static std::string kDuplicationEnvMasterClusterKey;
- const static std::string kDuplicationEnvMasterMetasKey;
- const static std::string kDuplicationEnvMasterAppNameKey;
+ const static std::string kEnvMasterClusterKey;
+ const static std::string kEnvMasterMetasKey;
+ const static std::string kEnvMasterAppNameKey;
+ const static std::string kEnvFollowerAppStatusKey;
+ const static std::string kEnvFollowerAppStatusCreating;
+ const static std::string kEnvFollowerAppStatusCreated;
};
USER_DEFINED_ENUM_FORMATTER(duplication_fail_mode::type)
USER_DEFINED_ENUM_FORMATTER(duplication_status::type)
+
} // namespace replication
} // namespace dsn
diff --git a/src/meta/app_env_validator.cpp b/src/meta/app_env_validator.cpp
index 06288cb61..bb59b43e4 100644
--- a/src/meta/app_env_validator.cpp
+++ b/src/meta/app_env_validator.cpp
@@ -26,6 +26,7 @@
#include <utility>
#include <vector>
+#include "common/duplication_common.h"
#include "common/replica_envs.h"
#include "http/http_status_code.h"
#include "utils/fmt_logging.h"
@@ -59,7 +60,7 @@ bool app_env_validator::validate_app_envs(const
std::map<std::string, std::strin
}
std::string hint_message;
if (!validate_app_env(key, value, hint_message)) {
- LOG_WARNING("app env '{}={}' is invalid, hint_message: {}", key,
value, hint_message);
+ LOG_ERROR("app env '{}={}' is invalid, hint_message: {}", key,
value, hint_message);
return false;
}
}
@@ -339,7 +340,12 @@ void app_env_validator::register_all_validators()
[](int64_t new_value) { return new_value == -1 || new_value >= 1;
}}},
{replica_envs::MANUAL_COMPACT_PERIODIC_BOTTOMMOST_LEVEL_COMPACTION,
mcblc},
{replica_envs::REPLICA_ACCESS_CONTROLLER_ALLOWED_USERS,
{ValueType::kString}},
- {replica_envs::REPLICA_ACCESS_CONTROLLER_RANGER_POLICIES,
{ValueType::kString}}};
+ {replica_envs::REPLICA_ACCESS_CONTROLLER_RANGER_POLICIES,
{ValueType::kString}},
+ {duplication_constants::kEnvMasterClusterKey, {ValueType::kString}},
+ {duplication_constants::kEnvMasterMetasKey, {ValueType::kString}},
+ {duplication_constants::kEnvMasterAppNameKey, {ValueType::kString}},
+ {duplication_constants::kEnvFollowerAppStatusKey,
{ValueType::kString}},
+ };
}
const std::unordered_map<app_env_validator::ValueType, std::string>
diff --git a/src/meta/duplication/meta_duplication_service.cpp
b/src/meta/duplication/meta_duplication_service.cpp
index 66cd2ed12..c557c890a 100644
--- a/src/meta/duplication/meta_duplication_service.cpp
+++ b/src/meta/duplication/meta_duplication_service.cpp
@@ -25,7 +25,7 @@
#include <string_view>
#include <type_traits>
-#include "common//duplication_common.h"
+#include "common/duplication_common.h"
#include "common/common.h"
#include "common/gpid.h"
#include "common/replication.codes.h"
@@ -421,7 +421,7 @@ void
meta_duplication_service::duplication_sync(duplication_sync_rpc rpc)
if (dup->status() == duplication_status::DS_PREPARE) {
create_follower_app_for_duplication(dup, app);
} else if (dup->status() == duplication_status::DS_APP) {
- check_follower_app_if_create_completed(dup);
+ mark_follower_app_created_for_duplication(dup, app);
}
}
@@ -463,6 +463,53 @@ void
meta_duplication_service::duplication_sync(duplication_sync_rpc rpc)
void meta_duplication_service::create_follower_app_for_duplication(
const std::shared_ptr<duplication_info> &dup, const
std::shared_ptr<app_state> &app)
+{
+ // The request of creating table might be issued to the follower cluster
many times,
+ // since some error might occurred while the follower table is being
created. For
+ // example, once the follower cluster could no connect to the master
cluster, the
+ // checkpoint would not be duplicated to the master cluster then the
follower table
+ // failed to be created; in this case, the master cluster would repeat
this request
+ // until the follower is created.
+ //
+ // To make this request idempotent, it would carry a status marking the
follower as
+ // creating by an environment variable. The follower cluster would also
store the
+ // status with the table as an env. As long as the status of the table is
creating,
+ // the follower cluster would accept the request even if it is repeated
(actually
+ // just ignore) and reply with ok. On the side of master cluster, it would
send the
+ // same request periodically as long as it is still at the status of
DS_PREPARE.
+ do_create_follower_app_for_duplication(
+ dup,
+ app,
+ duplication_constants::kEnvFollowerAppStatusCreating,
+ [this, dup](error_code err, configuration_create_app_response &&resp) {
+ on_follower_app_creating_for_duplication(dup, err,
std::move(resp));
+ });
+}
+
+void meta_duplication_service::mark_follower_app_created_for_duplication(
+ const std::shared_ptr<duplication_info> &dup, const
std::shared_ptr<app_state> &app)
+{
+ // Once the status of duplication has become DS_APP, the master cluster
would send
+ // another request to the follower cluster to mark the table as created.
If the
+ // table is at the status of creating, the follower cluster should accept
the request
+ // and update its status as created. Similarly, the master cluster could
send this
+ // request repeatedly and the follower would accept. The follower would
reject if it
+ // receives some invalid request(for example, a request that tries to
create the same
+ // table again.
+ do_create_follower_app_for_duplication(
+ dup,
+ app,
+ duplication_constants::kEnvFollowerAppStatusCreated,
+ [this, dup](error_code err, configuration_create_app_response &&resp) {
+ on_follower_app_created_for_duplication(dup, err, std::move(resp));
+ });
+}
+
+void meta_duplication_service::do_create_follower_app_for_duplication(
+ const std::shared_ptr<duplication_info> &dup,
+ const std::shared_ptr<app_state> &app,
+ const std::string &create_status,
+ std::function<void(error_code, configuration_create_app_response &&)>
create_callback)
{
configuration_create_app_request request;
request.app_name = dup->remote_app_name;
@@ -475,14 +522,14 @@ void
meta_duplication_service::create_follower_app_for_duplication(
// add envs for follower table, which will use it know itself is
`follower` and load master info
// - env map:
- // `kDuplicationEnvMasterClusterKey=>{master_cluster_name}`
- // `kDuplicationEnvMasterMetasKey=>{master_meta_list}`
-
request.options.envs.emplace(duplication_constants::kDuplicationEnvMasterClusterKey,
+ // `kEnvMasterClusterKey=>{master_cluster_name}`
+ // `kEnvMasterMetasKey=>{master_meta_list}`
+ request.options.envs.emplace(duplication_constants::kEnvMasterClusterKey,
get_current_dup_cluster_name());
-
request.options.envs.emplace(duplication_constants::kDuplicationEnvMasterMetasKey,
+ request.options.envs.emplace(duplication_constants::kEnvMasterMetasKey,
_meta_svc->get_meta_list_string());
-
request.options.envs.emplace(duplication_constants::kDuplicationEnvMasterAppNameKey,
- app->app_name);
+ request.options.envs.emplace(duplication_constants::kEnvMasterAppNameKey,
app->app_name);
+
request.options.envs.emplace(duplication_constants::kEnvFollowerAppStatusKey,
create_status);
host_port meta_servers;
meta_servers.assign_group(dup->remote_cluster_name.c_str());
@@ -490,47 +537,94 @@ void
meta_duplication_service::create_follower_app_for_duplication(
dsn::message_ex *msg = dsn::message_ex::create_request(RPC_CM_CREATE_APP);
dsn::marshall(msg, request);
- rpc::call(
- dsn::dns_resolver::instance().resolve_address(meta_servers),
- msg,
- _meta_svc->tracker(),
- [dup, this](error_code err, configuration_create_app_response &&resp)
mutable {
- FAIL_POINT_INJECT_NOT_RETURN_F("update_app_request_ok",
- [&err](std::string_view) -> void {
err = ERR_OK; });
- error_code create_err = err == ERR_OK ? resp.err : err;
- FAIL_POINT_INJECT_NOT_RETURN_F(
- "persist_dup_status_failed",
- [&create_err](std::string_view) -> void { create_err = ERR_OK;
});
+ LOG_INFO("send request to create follower app(cluster_name={},
app_name={}, status={}) "
+ "to trigger duplicate checkpoint: master_app_name={},
duplication_status={}",
+ dup->remote_cluster_name,
+ dup->remote_app_name,
+ create_status,
+ app->app_name,
+ duplication_status_to_string(dup->status()));
+
+ rpc::call(dsn::dns_resolver::instance().resolve_address(meta_servers),
+ msg,
+ _meta_svc->tracker(),
+ std::move(create_callback));
+}
- error_code update_err = ERR_NO_NEED_OPERATE;
- if (create_err == ERR_OK) {
- update_err = dup->alter_status(duplication_status::DS_APP);
- }
+void meta_duplication_service::on_follower_app_creating_for_duplication(
+ const std::shared_ptr<duplication_info> &dup,
+ error_code err,
+ configuration_create_app_response &&resp)
+{
+ FAIL_POINT_INJECT_NOT_RETURN_F("update_app_request_ok",
+ [&err](std::string_view) -> void { err =
ERR_OK; });
- FAIL_POINT_INJECT_F("persist_dup_status_failed",
- [](std::string_view) -> void { return; });
+ error_code create_err = err == ERR_OK ? resp.err : err;
+ FAIL_POINT_INJECT_NOT_RETURN_F(
+ "persist_dup_status_failed",
+ [&create_err](std::string_view) -> void { create_err = ERR_OK; });
- if (update_err == ERR_OK) {
- blob value = dup->to_json_blob();
- // Note: this function is `async`, it may not be persisted
completed
- // after executing, now using `_is_altering` to judge whether
`updating` or
- // `completed`, if `_is_altering`, dup->alter_status() will
return `ERR_BUSY`
-
_meta_svc->get_meta_storage()->set_data(std::string(dup->store_path),
- std::move(value),
- [dup]() {
dup->persist_status(); });
- } else {
- LOG_ERROR("create follower app[{}.{}] to trigger duplicate
checkpoint failed: "
- "duplication_status = {}, create_err = {},
update_err = {}",
- dup->remote_cluster_name,
- dup->remote_app_name,
- duplication_status_to_string(dup->status()),
- create_err,
- update_err);
- }
+ error_code update_err = ERR_NO_NEED_OPERATE;
+ if (create_err == ERR_OK) {
+ update_err = dup->alter_status(duplication_status::DS_APP);
+ }
+
+ FAIL_POINT_INJECT_F("persist_dup_status_failed", [](std::string_view) ->
void { return; });
+
+ if (update_err != ERR_OK) {
+ LOG_ERROR("create follower app(cluster_name={}, app_name={}) to
trigger duplicate "
+ "checkpoint failed: master_app_name={},
duplication_status={}, "
+ "create_err={}, update_err={}",
+ dup->remote_cluster_name,
+ dup->remote_app_name,
+ dup->app_name,
+ duplication_status_to_string(dup->status()),
+ create_err,
+ update_err);
+ return;
+ }
+
+ blob value = dup->to_json_blob();
+ // Note: this function is `async`, it may not be persisted completed
+ // after executing, now using `_is_altering` to judge whether `updating` or
+ // `completed`, if `_is_altering`, dup->alter_status() will return
`ERR_BUSY`
+ _meta_svc->get_meta_storage()->set_data(
+ std::string(dup->store_path), std::move(value), [dup]() {
+ dup->persist_status();
+ LOG_INFO("create follower app(cluster_name={}, app_name={}) to
trigger duplicate "
+ "checkpoint successfully: master_app_name={},
duplication_status={}",
+ dup->remote_cluster_name,
+ dup->remote_app_name,
+ dup->app_name,
+ duplication_status_to_string(dup->status()));
});
}
+void meta_duplication_service::on_follower_app_created_for_duplication(
+ const std::shared_ptr<duplication_info> &dup,
+ error_code err,
+ configuration_create_app_response &&resp)
+{
+ FAIL_POINT_INJECT_NOT_RETURN_F("on_follower_app_created",
[&err](std::string_view s) -> void {
+ err = error_code(s.data());
+ });
+
+ if (err != ERR_OK || resp.err != ERR_OK) {
+ LOG_ERROR("mark follower app(cluster_name={}, app_name={}) as created
failed: "
+ "master_app_name={}, duplication_status={}, callback_err={},
resp_err={}",
+ dup->remote_cluster_name,
+ dup->remote_app_name,
+ dup->app_name,
+ duplication_status_to_string(dup->status()),
+ err,
+ resp.err);
+ return;
+ }
+
+ check_follower_app_if_create_completed(dup);
+}
+
namespace {
// The format of `replica_state_str` is
"<has_primary>,<valid_secondaries>,<invalid_secondaries>":
@@ -599,6 +693,14 @@ void
meta_duplication_service::check_follower_app_if_create_completed(
dsn::message_ex *msg =
dsn::message_ex::create_request(RPC_CM_QUERY_PARTITION_CONFIG_BY_INDEX);
dsn::marshall(msg, meta_config_request);
+
+ LOG_INFO("send request to check if all replicas of creating follower
app(cluster_name={}, "
+ "app_name={}) are ready: master_app_name={},
duplication_status={}",
+ dup->remote_cluster_name,
+ dup->remote_app_name,
+ dup->app_name,
+ duplication_status_to_string(dup->status()));
+
rpc::call(
dsn::dns_resolver::instance().resolve_address(meta_servers),
msg,
@@ -653,23 +755,33 @@ void
meta_duplication_service::check_follower_app_if_create_completed(
FAIL_POINT_INJECT_F("persist_dup_status_failed",
[](std::string_view) -> void { return; });
- if (update_err == ERR_OK) {
- blob value = dup->to_json_blob();
- // Note: this function is `async`, it may not be persisted
completed
- // after executing, now using `_is_altering` to judge whether
`updating` or
- // `completed`, if `_is_altering`, dup->alter_status() will
return `ERR_BUSY`
-
_meta_svc->get_meta_storage()->set_data(std::string(dup->store_path),
- std::move(value),
- [dup]() {
dup->persist_status(); });
- } else {
- LOG_ERROR("query follower app[{}.{}] replica configuration
completed, result: "
- "duplication_status = {}, query_err = {}, update_err
= {}",
+ if (update_err != ERR_OK) {
+ LOG_ERROR("check all replicas of creating follower
app(cluster_name={}, "
+ "app_name={}): master_app_name={},
duplication_status={}, "
+ "query_err={}, update_err={}",
dup->remote_cluster_name,
dup->remote_app_name,
+ dup->app_name,
duplication_status_to_string(dup->status()),
query_err,
update_err);
+ return;
}
+
+ blob value = dup->to_json_blob();
+ // Note: this function is `async`, it may not be persisted
completed
+ // after executing, now using `_is_altering` to judge whether
`updating` or
+ // `completed`, if `_is_altering`, dup->alter_status() will return
`ERR_BUSY`
+ _meta_svc->get_meta_storage()->set_data(
+ std::string(dup->store_path), std::move(value), [dup]() {
+ dup->persist_status();
+ LOG_INFO("all replicas of follower app(cluster_name={},
app_name={}) "
+ "have been ready: master_app_name={},
duplication_status={}",
+ dup->remote_cluster_name,
+ dup->remote_app_name,
+ dup->app_name,
+ duplication_status_to_string(dup->status()));
+ });
});
}
diff --git a/src/meta/duplication/meta_duplication_service.h
b/src/meta/duplication/meta_duplication_service.h
index 3f06d6326..1cb700d7d 100644
--- a/src/meta/duplication/meta_duplication_service.h
+++ b/src/meta/duplication/meta_duplication_service.h
@@ -17,7 +17,8 @@
#pragma once
-#include <stdint.h>
+#include <cstdint>
+#include <functional>
#include <map>
#include <memory>
#include <string>
@@ -36,6 +37,7 @@ class host_port;
class zrwlock_nr;
namespace replication {
+class configuration_create_app_response;
class duplication_confirm_entry;
class duplication_query_request;
class duplication_query_response;
@@ -101,8 +103,37 @@ private:
int32_t partition_idx,
const duplication_confirm_entry
&confirm_entry);
+ // Send a request to the follower cluster to create a table for
duplication. This request
+ // is idempotent and is allowed to be sent multiple times as long as the
duplication is at
+ // the status of DS_PREPARE.
void create_follower_app_for_duplication(const
std::shared_ptr<duplication_info> &dup,
const std::shared_ptr<app_state>
&app);
+
+ // Send a request to the follower cluster to mark a table as created for
duplication. This
+ // request is idempotent and is allowed to be sent multiple times as long
as the duplication
+ // is at the status of DS_APP.
+ void mark_follower_app_created_for_duplication(const
std::shared_ptr<duplication_info> &dup,
+ const
std::shared_ptr<app_state> &app);
+
+ // Send a request to the follower cluster to create a table or mark it as
some specific
+ // status.
+ void do_create_follower_app_for_duplication(
+ const std::shared_ptr<duplication_info> &dup,
+ const std::shared_ptr<app_state> &app,
+ const std::string &create_status,
+ std::function<void(error_code, configuration_create_app_response &&)>
create_callback);
+
+ // Callback for the response of creaing a new table.
+ void on_follower_app_creating_for_duplication(const
std::shared_ptr<duplication_info> &dup,
+ error_code err,
+
configuration_create_app_response &&resp);
+
+ // Callback for the response of marking a table as created.
+ void on_follower_app_created_for_duplication(const
std::shared_ptr<duplication_info> &dup,
+ error_code err,
+
configuration_create_app_response &&resp);
+
+ // Check if the whole follower table(including all of its partitions and
replicas) is ready.
void check_follower_app_if_create_completed(const
std::shared_ptr<duplication_info> &dup);
// Get zk path for duplication.
diff --git a/src/meta/server_state.cpp b/src/meta/server_state.cpp
index 68bc78618..8d7c97f2f 100644
--- a/src/meta/server_state.cpp
+++ b/src/meta/server_state.cpp
@@ -126,8 +126,17 @@ DSN_DEFINE_int32(meta_server,
DSN_DECLARE_bool(recover_from_replica_server);
-namespace dsn {
-namespace replication {
+namespace dsn::replication {
+
+// Reply to the client with specified response.
+#define REPLY_TO_CLIENT(msg, response)
\
+ _meta_svc->reply_data(msg, response);
\
+ msg->release_ref()
+
+// Reply to the client with specified response, and return from current
function.
+#define REPLY_TO_CLIENT_AND_RETURN(msg, response)
\
+ REPLY_TO_CLIENT(msg, response);
\
+ return
static const char *lock_state = "lock";
static const char *unlock_state = "unlock";
@@ -207,6 +216,31 @@ int server_state::count_staging_app()
return ans;
}
+// Create a new variable of `configuration_create_app_response` and assign it
with specified
+// error code.
+#define INIT_CREATE_APP_RESPONSE_WITH_ERR(response, err_code)
\
+ configuration_create_app_response response;
\
+ response.err = err_code
+
+// Create a new variable of `configuration_create_app_response` and assign it
with ERR_OK
+// and table id.
+#define INIT_CREATE_APP_RESPONSE_WITH_OK(response, app_id)
\
+ configuration_create_app_response response;
\
+ response.err = dsn::ERR_OK;
\
+ response.appid = app_id;
+
+// Reply to the client with a newly created failed
`configuration_create_app_response` and return
+// from current function.
+#define FAIL_CREATE_APP_RESPONSE(msg, response, err_code)
\
+ INIT_CREATE_APP_RESPONSE_WITH_ERR(response, err_code);
\
+ REPLY_TO_CLIENT_AND_RETURN(msg, response)
+
+// Reply to the client with a newly created successful
`configuration_create_app_response` and
+// return from current function.
+#define SUCC_CREATE_APP_RESPONSE(msg, response, app_id)
\
+ INIT_CREATE_APP_RESPONSE_WITH_OK(response, app_id);
\
+ REPLY_TO_CLIENT_AND_RETURN(msg, response)
+
void server_state::transition_staging_state(std::shared_ptr<app_state> &app)
{
#define send_response(meta, msg, response_data)
\
@@ -221,8 +255,7 @@ void
server_state::transition_staging_state(std::shared_ptr<app_state> &app)
app_status::type old_status = app->status;
if (app->status == app_status::AS_CREATING) {
app->status = app_status::AS_AVAILABLE;
- configuration_create_app_response resp;
- resp.err = dsn::ERR_OK;
+ INIT_CREATE_APP_RESPONSE_WITH_ERR(resp, dsn::ERR_OK);
resp.appid = app->app_id;
send_response(_meta_svc, app->helpers->pending_response, resp);
} else if (app->status == app_status::AS_DROPPING) {
@@ -1081,25 +1114,22 @@ void
server_state::do_app_create(std::shared_ptr<app_state> &app)
void server_state::create_app(dsn::message_ex *msg)
{
configuration_create_app_request request;
- configuration_create_app_response response;
- std::shared_ptr<app_state> app;
- bool will_create_app = false;
dsn::unmarshall(msg, request);
- const auto &duplication_env_iterator =
-
request.options.envs.find(duplication_constants::kDuplicationEnvMasterClusterKey);
+ const auto &master_cluster =
+ request.options.envs.find(duplication_constants::kEnvMasterClusterKey);
+ bool duplicating = master_cluster != request.options.envs.end();
LOG_INFO("create app request, name({}), type({}), partition_count({}),
replica_count({}), "
"duplication({})",
request.app_name,
request.options.app_type,
request.options.partition_count,
request.options.replica_count,
- duplication_env_iterator == request.options.envs.end()
- ? "false"
- : fmt::format(
- "{}.{}",
-
request.options.envs[duplication_constants::kDuplicationEnvMasterClusterKey],
- request.app_name));
+ duplicating
+ ? fmt::format("{}.{}",
+
request.options.envs[duplication_constants::kEnvMasterClusterKey],
+ request.app_name)
+ : "false");
auto option_match_check = [](const create_app_options &opt, const
app_state &exist_app) {
return opt.partition_count == exist_app.partition_count &&
@@ -1111,71 +1141,234 @@ void server_state::create_app(dsn::message_ex *msg)
auto level = _meta_svc->get_function_level();
if (level <= meta_function_level::fl_freezed) {
LOG_ERROR("current meta function level is freezed, since there are too
few alive nodes");
- response.err = ERR_STATE_FREEZED;
- will_create_app = false;
- } else if (request.options.partition_count <= 0 ||
-
!validate_target_max_replica_count(request.options.replica_count)) {
- response.err = ERR_INVALID_PARAMETERS;
- will_create_app = false;
- } else if (!_app_env_validator.validate_app_envs(request.options.envs)) {
- response.err = ERR_INVALID_PARAMETERS;
- will_create_app = false;
- } else {
- zauto_write_lock l(_lock);
- app = get_app(request.app_name);
- if (nullptr != app) {
- switch (app->status) {
- case app_status::AS_AVAILABLE:
- if (!request.options.success_if_exist) {
- response.err = ERR_APP_EXIST;
- } else if (!option_match_check(request.options, *app)) {
- response.err = ERR_INVALID_PARAMETERS;
- } else {
- response.err = ERR_OK;
- response.appid = app->app_id;
+ FAIL_CREATE_APP_RESPONSE(msg, response, ERR_STATE_FREEZED);
+ }
+
+ if (request.options.partition_count <= 0) {
+ LOG_ERROR("partition_count({}) is invalid",
request.options.partition_count);
+ FAIL_CREATE_APP_RESPONSE(msg, response, ERR_INVALID_PARAMETERS);
+ }
+
+ if (!validate_target_max_replica_count(request.options.replica_count)) {
+ FAIL_CREATE_APP_RESPONSE(msg, response, ERR_INVALID_PARAMETERS);
+ }
+
+ if (!_app_env_validator.validate_app_envs(request.options.envs)) {
+ FAIL_CREATE_APP_RESPONSE(msg, response, ERR_INVALID_PARAMETERS);
+ }
+
+ zauto_write_lock l(_lock);
+
+ auto app = get_app(request.app_name);
+ if (nullptr != app) {
+ configuration_create_app_response response;
+
+ switch (app->status) {
+ case app_status::AS_AVAILABLE:
+ if (!request.options.success_if_exist) {
+ if (duplicating) {
+ process_create_follower_app_status(msg, request,
master_cluster->second, app);
+ return;
}
- break;
- case app_status::AS_CREATING:
- case app_status::AS_RECALLING:
- response.err = ERR_BUSY_CREATING;
- break;
- case app_status::AS_DROPPING:
- response.err = ERR_BUSY_DROPPING;
- break;
- default:
- break;
+
+ response.err = ERR_APP_EXIST;
+ } else if (!option_match_check(request.options, *app)) {
+ response.err = ERR_INVALID_PARAMETERS;
+ } else {
+ response.err = ERR_OK;
+ response.appid = app->app_id;
}
- } else {
- will_create_app = true;
-
- app_info info;
- info.app_id = next_app_id();
- info.app_name = request.app_name;
- info.app_type = request.options.app_type;
- info.envs = std::move(request.options.envs);
- info.is_stateful = request.options.is_stateful;
- info.max_replica_count = request.options.replica_count;
- info.partition_count = request.options.partition_count;
- info.status = app_status::AS_CREATING;
- info.create_second = dsn_now_ms() / 1000;
- info.init_partition_count = request.options.partition_count;
-
- app = app_state::create(info);
- app->helpers->pending_response = msg;
- app->helpers->partitions_in_progress.store(info.partition_count);
+ break;
+ case app_status::AS_CREATING:
+ case app_status::AS_RECALLING:
+ response.err = ERR_BUSY_CREATING;
+ break;
+ case app_status::AS_DROPPING:
+ response.err = ERR_BUSY_DROPPING;
+ break;
+ default:
+ break;
+ }
- _all_apps.emplace(app->app_id, app);
- _exist_apps.emplace(request.app_name, app);
- _table_metric_entities.create_entity(app->app_id,
app->partition_count);
+ REPLY_TO_CLIENT_AND_RETURN(msg, response);
+ }
+
+ app_info info;
+ info.app_id = next_app_id();
+ info.app_name = request.app_name;
+ info.app_type = request.options.app_type;
+ info.envs = std::move(request.options.envs);
+ info.is_stateful = request.options.is_stateful;
+ info.max_replica_count = request.options.replica_count;
+ info.partition_count = request.options.partition_count;
+ info.status = app_status::AS_CREATING;
+ info.create_second = static_cast<int64_t>(dsn_now_s());
+ info.init_partition_count = request.options.partition_count;
+
+ app = app_state::create(info);
+ app->helpers->pending_response = msg;
+ app->helpers->partitions_in_progress.store(info.partition_count);
+
+ _all_apps.emplace(app->app_id, app);
+ _exist_apps.emplace(request.app_name, app);
+ _table_metric_entities.create_entity(app->app_id, app->partition_count);
+
+ do_app_create(app);
+}
+
+// It is idempotent for the repeated requests.
+#define SUCC_IDEMPOTENT_CREATE_FOLLOWER_APP_STATUS()
\
+ LOG_INFO("repeated request that updates env {} of the follower app from {}
to {}, " \
+ "just ignore: app_name={}, app_id={}",
\
+ duplication_constants::kEnvFollowerAppStatusKey,
\
+ my_status->second,
\
+ req_status->second,
\
+ app->app_name,
\
+ app->app_id);
\
+ SUCC_CREATE_APP_RESPONSE(msg, response, app->app_id)
+
+// Failed due to invalid creating status.
+#define FAIL_UNDEFINED_CREATE_FOLLOWER_APP_STATUS(val, desc)
\
+ LOG_ERROR("undefined value({}) of env {} in the {}: app_name={},
app_id={}", \
+ val,
\
+ duplication_constants::kEnvFollowerAppStatusKey,
\
+ desc,
\
+ app->app_name,
\
+ app->app_id);
\
+ FAIL_CREATE_APP_RESPONSE(msg, response, ERR_INVALID_PARAMETERS)
+
+void server_state::process_create_follower_app_status(
+ message_ex *msg,
+ const configuration_create_app_request &request,
+ const std::string &req_master_cluster,
+ std::shared_ptr<app_state> &app)
+{
+ const auto &my_master_cluster =
app->envs.find(duplication_constants::kEnvMasterClusterKey);
+ if (my_master_cluster == app->envs.end() || my_master_cluster->second !=
req_master_cluster) {
+ // The source cluster is not matched.
+ LOG_ERROR("env {} are not matched between the request({}) and the
follower "
+ "app({}): app_name={}, app_id={}",
+ duplication_constants::kEnvMasterClusterKey,
+ req_master_cluster,
+ my_master_cluster == app->envs.end() ? "<nil>" :
my_master_cluster->second,
+ app->app_name,
+ app->app_id);
+ FAIL_CREATE_APP_RESPONSE(msg, response, ERR_APP_EXIST);
+ }
+
+ const auto &req_status =
+
request.options.envs.find(duplication_constants::kEnvFollowerAppStatusKey);
+ if (req_status == request.options.envs.end()) {
+ // Still reply with ERR_APP_EXIST to the master cluster of old
versions.
+ LOG_ERROR("no env {} in the request: app_name={}, app_id={}",
+ duplication_constants::kEnvFollowerAppStatusKey,
+ app->app_name,
+ app->app_id);
+ FAIL_CREATE_APP_RESPONSE(msg, response, ERR_APP_EXIST);
+ }
+
+ const auto &my_status =
app->envs.find(duplication_constants::kEnvFollowerAppStatusKey);
+ if (my_status == app->envs.end()) {
+ // Since currently this table have been AS_AVAILABLE, it should have
the env of
+ // creating status.
+ LOG_ERROR("no env {} in the follower app: app_name={}, app_id={}",
+ duplication_constants::kEnvFollowerAppStatusKey,
+ app->app_name,
+ app->app_id);
+ FAIL_CREATE_APP_RESPONSE(msg, response, ERR_INVALID_STATE);
+ return;
+ }
+
+ if (my_status->second ==
duplication_constants::kEnvFollowerAppStatusCreating) {
+ if (req_status->second ==
duplication_constants::kEnvFollowerAppStatusCreating) {
+ SUCC_IDEMPOTENT_CREATE_FOLLOWER_APP_STATUS();
}
+
+ if (req_status->second ==
duplication_constants::kEnvFollowerAppStatusCreated) {
+ // Mark the status as created both on the remote storage and local
memory.
+ update_create_follower_app_status(msg,
+
duplication_constants::kEnvFollowerAppStatusCreating,
+
duplication_constants::kEnvFollowerAppStatusCreated,
+ app);
+ return;
+ }
+
+ FAIL_UNDEFINED_CREATE_FOLLOWER_APP_STATUS(req_status->second,
"request");
}
- if (will_create_app) {
- do_app_create(app);
- } else {
- _meta_svc->reply_data(msg, response);
- msg->release_ref();
+ if (my_status->second ==
duplication_constants::kEnvFollowerAppStatusCreated) {
+ if (req_status->second ==
duplication_constants::kEnvFollowerAppStatusCreating) {
+ // The status of the duplication should have been DS_APP since the
follower app
+ // has been marked as created. Thus, the master cluster should
never send the
+ // request with creating status again.
+ LOG_ERROR("the master cluster should never send the request with
env {} valued {} "
+ "again since it has been {} in the follower app:
app_name={}, app_id={}",
+ duplication_constants::kEnvFollowerAppStatusKey,
+ req_status->second,
+ my_status->second,
+ app->app_name,
+ app->app_id);
+ FAIL_CREATE_APP_RESPONSE(msg, response, ERR_APP_EXIST);
+ }
+
+ if (req_status->second ==
duplication_constants::kEnvFollowerAppStatusCreated) {
+ SUCC_IDEMPOTENT_CREATE_FOLLOWER_APP_STATUS();
+ }
+
+ FAIL_UNDEFINED_CREATE_FOLLOWER_APP_STATUS(req_status->second,
"request");
}
+
+ // Some undefined creating status from the target table.
+ FAIL_UNDEFINED_CREATE_FOLLOWER_APP_STATUS(my_status->second, "follower
app");
+}
+
+#undef FAIL_UNDEFINED_CREATE_FOLLOWER_APP_STATUS
+#undef SUCC_IDEMPOTENT_CREATE_FOLLOWER_APP_STATUS
+
+void server_state::update_create_follower_app_status(message_ex *msg,
+ const std::string
&old_status,
+ const std::string
&new_status,
+
std::shared_ptr<app_state> &app)
+{
+ app_info ainfo = *app;
+ ainfo.envs[duplication_constants::kEnvFollowerAppStatusKey] = new_status;
+ auto app_path = get_app_path(*app);
+
+ LOG_INFO("ready to update env {} of follower app from {} to {},
app_name={}, app_id={}, ",
+ duplication_constants::kEnvFollowerAppStatusKey,
+ old_status,
+ new_status,
+ app->app_name,
+ app->app_id);
+
+ do_update_app_info(
+ app_path, ainfo, [this, msg, old_status, new_status, app](error_code
ec) mutable {
+ {
+ zauto_write_lock l(_lock);
+
+ if (ec != ERR_OK) {
+ LOG_ERROR("failed to update remote env of creating
follower app status: "
+ "error_code={}, app_name={}, app_id={}, {}={} =>
{}",
+ ec,
+ app->app_name,
+ app->app_id,
+ duplication_constants::kEnvFollowerAppStatusKey,
+ old_status,
+ new_status);
+ FAIL_CREATE_APP_RESPONSE(msg, response, ec);
+ }
+
+ app->envs[duplication_constants::kEnvFollowerAppStatusKey] =
new_status;
+ LOG_INFO("both remote and local env of creating follower app
status have been "
+ "updated successfully: app_name={}, app_id={}, {}={}
=> {}",
+ app->app_name,
+ app->app_id,
+ duplication_constants::kEnvFollowerAppStatusKey,
+ old_status,
+ new_status);
+ SUCC_CREATE_APP_RESPONSE(msg, response, app->app_id);
+ }
+ });
}
void server_state::do_app_drop(std::shared_ptr<app_state> &app)
@@ -1255,12 +1448,12 @@ void server_state::drop_app(dsn::message_ex *msg)
}
}
}
- if (do_dropping) {
- do_app_drop(app);
- } else {
- _meta_svc->reply_data(msg, response);
- msg->release_ref();
+
+ if (!do_dropping) {
+ REPLY_TO_CLIENT_AND_RETURN(msg, response);
}
+
+ do_app_drop(app);
}
void server_state::rename_app(configuration_rename_app_rpc rpc)
@@ -1404,10 +1597,9 @@ void server_state::recall_app(dsn::message_ex *msg)
}
if (!do_recalling) {
- _meta_svc->reply_data(msg, response);
- msg->release_ref();
- return;
+ REPLY_TO_CLIENT_AND_RETURN(msg, response);
}
+
do_app_recall(target_app);
}
@@ -1743,8 +1935,7 @@ void
server_state::on_update_configuration_on_remote_reply(
configuration_update_response resp;
resp.err = ERR_OK;
resp.config = config_request->config;
- _meta_svc->reply_data(cc.msg, resp);
- cc.msg->release_ref();
+ REPLY_TO_CLIENT(cc.msg, resp);
cc.msg = nullptr;
}
@@ -2031,17 +2222,16 @@ void server_state::on_update_configuration(
}
if (response.err != ERR_IO_PENDING) {
- _meta_svc->reply_data(msg, response);
- msg->release_ref();
- } else {
- CHECK(config_status::not_pending == cc.stage,
- "invalid config status, cc.stage = {}",
- enum_to_string(cc.stage));
- cc.stage = config_status::pending_remote_sync;
- cc.pending_sync_request = cfg_request;
- cc.msg = msg;
- cc.pending_sync_task = update_configuration_on_remote(cfg_request);
+ REPLY_TO_CLIENT_AND_RETURN(msg, response);
}
+
+ CHECK(config_status::not_pending == cc.stage,
+ "invalid config status, cc.stage = {}",
+ enum_to_string(cc.stage));
+ cc.stage = config_status::pending_remote_sync;
+ cc.pending_sync_request = cfg_request;
+ cc.msg = msg;
+ cc.pending_sync_task = update_configuration_on_remote(cfg_request);
}
void server_state::on_partition_node_dead(std::shared_ptr<app_state> &app,
@@ -4055,5 +4245,12 @@ void
server_state::recover_app_max_replica_count(std::shared_ptr<app_state> &app
&tracker);
}
-} // namespace replication
-} // namespace dsn
+#undef SUCC_CREATE_APP_RESPONSE
+#undef FAIL_CREATE_APP_RESPONSE
+#undef INIT_CREATE_APP_RESPONSE_WITH_OK
+#undef INIT_CREATE_APP_RESPONSE_WITH_ERR
+
+#undef REPLY_TO_CLIENT_AND_RETURN
+#undef REPLY_TO_CLIENT
+
+} // namespace dsn::replication
diff --git a/src/meta/server_state.h b/src/meta/server_state.h
index a787328f7..ea9a418ba 100644
--- a/src/meta/server_state.h
+++ b/src/meta/server_state.h
@@ -44,21 +44,22 @@
#include "dsn.layer2_types.h"
#include "meta/meta_rpc_types.h"
#include "meta_data.h"
+#include "table_metrics.h"
#include "task/task.h"
#include "task/task_tracker.h"
-#include "table_metrics.h"
#include "utils/error_code.h"
#include "utils/zlocks.h"
namespace dsn {
class blob;
class command_deregister;
-class message_ex;
class host_port;
+class message_ex;
namespace replication {
class configuration_balancer_request;
class configuration_balancer_response;
+class configuration_create_app_request;
class configuration_list_apps_request;
class configuration_list_apps_response;
class configuration_proposal_action;
@@ -258,6 +259,21 @@ private:
bool skip_lost_partitions,
std::string &hint_message);
+ // Process the status carried in the environment variables of creating
table request while
+ // the table is at the status of AS_AVAILABLE, to update remote and local
states and reply
+ // to the master cluster.
+ void process_create_follower_app_status(message_ex *msg,
+ const
configuration_create_app_request &request,
+ const std::string
&req_master_cluster,
+ std::shared_ptr<app_state> &app);
+
+ // Update the meta data with the new creating status both on the remote
storage and local
+ // memory.
+ void update_create_follower_app_status(message_ex *msg,
+ const std::string &old_status,
+ const std::string &new_status,
+ std::shared_ptr<app_state> &app);
+
void do_app_create(std::shared_ptr<app_state> &app);
void do_app_drop(std::shared_ptr<app_state> &app);
void do_app_recall(std::shared_ptr<app_state> &app);
diff --git a/src/meta/test/meta_app_operation_test.cpp
b/src/meta/test/meta_app_operation_test.cpp
index f8b399d4c..9956ec31e 100644
--- a/src/meta/test/meta_app_operation_test.cpp
+++ b/src/meta/test/meta_app_operation_test.cpp
@@ -16,7 +16,8 @@
// under the License.
#include <fmt/core.h>
-#include <stdint.h>
+#include <cstdint>
+#include <algorithm>
#include <functional>
#include <iostream>
#include <map>
@@ -26,6 +27,7 @@
#include <utility>
#include <vector>
+#include "common/duplication_common.h"
#include "common/gpid.h"
#include "common/json_helper.h"
#include "common/replica_envs.h"
@@ -62,7 +64,7 @@ namespace replication {
class meta_app_operation_test : public meta_test_base
{
public:
- meta_app_operation_test() {}
+ meta_app_operation_test() = default;
error_code create_app_test(int32_t partition_count,
int32_t replica_count,
@@ -320,58 +322,43 @@ public:
tracker.wait_outstanding_tasks();
}
+ void verify_app_envs(const std::string &app_name,
+ const std::map<std::string, std::string>
&expected_envs)
+ {
+ auto app = find_app(app_name);
+ CHECK(app, "app({}) does not exist", app_name);
+
+ auto app_path = _ss->get_app_path(*app);
+
+ dsn::task_tracker tracker;
+ _ms->get_remote_storage()->get_data(
+ app_path,
+ LPC_META_CALLBACK,
+ [app_name, expected_envs, app](error_code ec, const blob &value) {
+ ASSERT_EQ(ERR_OK, ec);
+
+ app_info ainfo;
+ dsn::json::json_forwarder<app_info>::decode(value, ainfo);
+
+ ASSERT_EQ(app_name, app->app_name);
+ ASSERT_EQ(app_name, ainfo.app_name);
+ ASSERT_EQ(app->app_id, ainfo.app_id);
+ ASSERT_EQ(expected_envs, app->envs);
+ ASSERT_EQ(expected_envs, ainfo.envs);
+ },
+ &tracker);
+ tracker.wait_outstanding_tasks();
+ }
+
const std::string APP_NAME = "app_operation_test";
const std::string OLD_APP_NAME = "old_app_operation";
+ const std::string DUP_MASTER_APP_NAME = "dup_master_test";
+ const std::string DUP_FOLLOWER_APP_NAME = "dup_follower_test";
};
TEST_F(meta_app_operation_test, create_app)
{
// Test cases: (assert min_allowed_replica_count <=
max_allowed_replica_count)
- // - wrong partition_count (< 0)
- // - wrong partition_count (= 0)
- // - wrong replica_count (< 0)
- // - wrong replica_count (= 0)
- // - wrong replica_count (> max_allowed_replica_count > alive_node_count)
- // - wrong replica_count (> alive_node_count > max_allowed_replica_count)
- // - wrong replica_count (> alive_node_count = max_allowed_replica_count)
- // - wrong replica_count (= max_allowed_replica_count, and >
alive_node_count)
- // - wrong replica_count (< max_allowed_replica_count, and >
alive_node_count)
- // - wrong replica_count (= alive_node_count, and >
max_allowed_replica_count)
- // - wrong replica_count (< alive_node_count, and >
max_allowed_replica_count)
- // - valid replica_count (= max_allowed_replica_count, and =
alive_node_count)
- // - valid replica_count (= max_allowed_replica_count, and <
alive_node_count)
- // - valid replica_count (< max_allowed_replica_count, and =
alive_node_count)
- // - valid replica_count (< max_allowed_replica_count < alive_node_count)
- // - valid replica_count (< alive_node_count < max_allowed_replica_count)
- // - valid replica_count (< alive_node_count = max_allowed_replica_count)
- // - wrong replica_count (< min_allowed_replica_count < alive_node_count)
- // - wrong replica_count (< alive_node_count < min_allowed_replica_count)
- // - wrong replica_count (< min_allowed_replica_count = alive_node_count)
- // - wrong replica_count (< min_allowed_replica_count, and >
alive_node_count)
- // - wrong replica_count (< min_allowed_replica_count, and =
alive_node_count)
- // - wrong replica_count (= min_allowed_replica_count, and >
alive_node_count)
- // - valid replica_count (= min_allowed_replica_count, and <
alive_node_count)
- // - cluster freezed (alive_node_count = 0)
- // - cluster freezed (alive_node_count = 1 <
min_live_node_count_for_unfreeze)
- // - cluster freezed (alive_node_count = 2 <
min_live_node_count_for_unfreeze)
- // - cluster not freezed (alive_node_count =
min_live_node_count_for_unfreeze)
- // - create succeed with single-replica
- // - create succeed with double-replica
- // - create app succeed
- // - create failed with table existed
- // - wrong app_status creating
- // - wrong app_status recalling
- // - wrong app_status dropping
- // - create succeed with app_status dropped
- // - create succeed with success_if_exist=true
- // - wrong rocksdb.num_levels (< 1)
- // - wrong rocksdb.num_levels (> 10)
- // - wrong rocksdb.num_levels (non-digital character)
- // - create app with rocksdb.num_levels (= 5) succeed
- // - wrong rocksdb.write_buffer_size (< (16<<20))
- // - wrong rocksdb.write_buffer_size (> (512<<20))
- // - wrong rocksdb.write_buffer_size (non-digital character)
- // - create app with rocksdb.write_buffer_size (= (32<<20)) succeed
struct create_test
{
std::string app_name;
@@ -385,43 +372,81 @@ TEST_F(meta_app_operation_test, create_app)
error_code expected_err;
std::map<std::string, std::string> envs = {};
} tests[] = {
+ // Wrong partition_count (< 0).
{APP_NAME, -1, 3, 2, 3, 1, false, app_status::AS_INVALID,
ERR_INVALID_PARAMETERS},
+ // Wrong partition_count (= 0).
{APP_NAME, 0, 3, 2, 3, 1, false, app_status::AS_INVALID,
ERR_INVALID_PARAMETERS},
+ // Wrong replica_count (< 0).
{APP_NAME, 4, -1, 1, 3, 1, false, app_status::AS_INVALID,
ERR_INVALID_PARAMETERS},
+ // Wrong replica_count (= 0).
{APP_NAME, 4, 0, 1, 3, 1, false, app_status::AS_INVALID,
ERR_INVALID_PARAMETERS},
+ // Wrong replica_count (> max_allowed_replica_count >
alive_node_count).
{APP_NAME, 4, 6, 2, 4, 1, false, app_status::AS_INVALID,
ERR_INVALID_PARAMETERS},
+ // Wrong replica_count (> alive_node_count >
max_allowed_replica_count).
{APP_NAME, 4, 7, 2, 6, 1, false, app_status::AS_INVALID,
ERR_INVALID_PARAMETERS},
+ // Wrong replica_count (> alive_node_count =
max_allowed_replica_count).
{APP_NAME, 4, 6, 2, 5, 1, false, app_status::AS_INVALID,
ERR_INVALID_PARAMETERS},
+ // Wrong replica_count (= max_allowed_replica_count, and >
alive_node_count).
{APP_NAME, 4, 5, 2, 4, 1, false, app_status::AS_INVALID,
ERR_INVALID_PARAMETERS},
+ // Wrong replica_count (< max_allowed_replica_count, and >
alive_node_count).
{APP_NAME, 4, 4, 2, 3, 1, false, app_status::AS_INVALID,
ERR_INVALID_PARAMETERS},
+ // Wrong replica_count (= alive_node_count, and >
max_allowed_replica_count).
{APP_NAME, 4, 6, 2, 6, 1, false, app_status::AS_INVALID,
ERR_INVALID_PARAMETERS},
+ // Wrong replica_count (< alive_node_count, and >
max_allowed_replica_count).
{APP_NAME, 4, 6, 2, 7, 1, false, app_status::AS_INVALID,
ERR_INVALID_PARAMETERS},
+ // Valid replica_count (= max_allowed_replica_count, and =
alive_node_count).
{APP_NAME + "_1", 4, 5, 2, 5, 1, false, app_status::AS_INVALID,
ERR_OK},
+ // Valid replica_count (= max_allowed_replica_count, and <
alive_node_count).
{APP_NAME + "_2", 4, 5, 2, 6, 1, false, app_status::AS_INVALID,
ERR_OK},
+ // Valid replica_count (< max_allowed_replica_count, and =
alive_node_count).
{APP_NAME + "_3", 4, 4, 2, 4, 1, false, app_status::AS_INVALID,
ERR_OK},
+ // Valid replica_count (< max_allowed_replica_count <
alive_node_count).
{APP_NAME + "_4", 4, 4, 2, 6, 1, false, app_status::AS_INVALID,
ERR_OK},
+ // Valid replica_count (< alive_node_count <
max_allowed_replica_count).
{APP_NAME + "_5", 4, 3, 2, 4, 1, false, app_status::AS_INVALID,
ERR_OK},
+ // Valid replica_count (< alive_node_count =
max_allowed_replica_count).
{APP_NAME + "_6", 4, 4, 2, 5, 1, false, app_status::AS_INVALID,
ERR_OK},
+ // Wrong replica_count (< min_allowed_replica_count <
alive_node_count).
{APP_NAME, 4, 3, 2, 5, 4, false, app_status::AS_INVALID,
ERR_INVALID_PARAMETERS},
+ // Wrong replica_count (< alive_node_count <
min_allowed_replica_count).
{APP_NAME, 4, 3, 2, 4, 5, false, app_status::AS_INVALID,
ERR_INVALID_PARAMETERS},
+ // Wrong replica_count (< min_allowed_replica_count =
alive_node_count).
{APP_NAME, 4, 3, 2, 4, 4, false, app_status::AS_INVALID,
ERR_INVALID_PARAMETERS},
+ // Wrong replica_count (< min_allowed_replica_count, and >
alive_node_count).
{APP_NAME, 4, 3, 2, 2, 4, false, app_status::AS_INVALID,
ERR_INVALID_PARAMETERS},
+ // Wrong replica_count (< min_allowed_replica_count, and =
alive_node_count).
{APP_NAME, 4, 3, 2, 3, 4, false, app_status::AS_INVALID,
ERR_INVALID_PARAMETERS},
+ // Wrong replica_count (= min_allowed_replica_count, and >
alive_node_count).
{APP_NAME, 4, 4, 2, 3, 4, false, app_status::AS_INVALID,
ERR_INVALID_PARAMETERS},
+ // Valid replica_count (= min_allowed_replica_count, and <
alive_node_count).
{APP_NAME + "_7", 4, 3, 2, 4, 3, false, app_status::AS_INVALID,
ERR_OK},
+ // Cluster freezed (alive_node_count = 0).
{APP_NAME, 4, 1, 1, 0, 1, false, app_status::AS_INVALID,
ERR_STATE_FREEZED},
+ // Cluster freezed (alive_node_count = 1 <
min_live_node_count_for_unfreeze).
{APP_NAME, 4, 2, 2, 1, 1, false, app_status::AS_INVALID,
ERR_STATE_FREEZED},
+ // Cluster freezed (alive_node_count = 2 <
min_live_node_count_for_unfreeze).
{APP_NAME, 4, 3, 3, 2, 1, false, app_status::AS_INVALID,
ERR_STATE_FREEZED},
+ // Cluster not freezed (alive_node_count =
min_live_node_count_for_unfreeze).
{APP_NAME + "_8", 4, 3, 3, 3, 1, false, app_status::AS_INVALID,
ERR_OK},
+ // Create succeed with single-replica.
{APP_NAME + "_9", 4, 1, 1, 1, 1, false, app_status::AS_INVALID,
ERR_OK},
+ // Create succeed with double-replica.
{APP_NAME + "_10", 4, 2, 1, 2, 2, false, app_status::AS_INVALID,
ERR_OK},
+ // Create app succeed.
{APP_NAME, 4, 3, 2, 3, 3, false, app_status::AS_INVALID, ERR_OK},
+ // Create failed with table existed.
{APP_NAME, 4, 3, 2, 3, 3, false, app_status::AS_INVALID,
ERR_APP_EXIST},
+ // Wrong app_status creating.
{APP_NAME, 4, 3, 2, 3, 3, false, app_status::AS_CREATING,
ERR_BUSY_CREATING},
+ // Wrong app_status recalling.
{APP_NAME, 4, 3, 2, 3, 3, false, app_status::AS_RECALLING,
ERR_BUSY_CREATING},
+ // Wrong app_status dropping.
{APP_NAME, 4, 3, 2, 3, 3, false, app_status::AS_DROPPING,
ERR_BUSY_DROPPING},
+ // Create succeed with app_status dropped.
{APP_NAME, 4, 3, 2, 3, 3, false, app_status::AS_DROPPED, ERR_OK},
+ // Create succeed with success_if_exist=true.
{APP_NAME, 4, 3, 2, 3, 3, true, app_status::AS_INVALID, ERR_OK},
+ // Wrong rocksdb.num_levels (< 1).
{APP_NAME,
4,
3,
@@ -432,6 +457,7 @@ TEST_F(meta_app_operation_test, create_app)
app_status::AS_INVALID,
ERR_INVALID_PARAMETERS,
{{"rocksdb.num_levels", "0"}}},
+ // Wrong rocksdb.num_levels (> 10).
{APP_NAME,
4,
3,
@@ -442,6 +468,7 @@ TEST_F(meta_app_operation_test, create_app)
app_status::AS_INVALID,
ERR_INVALID_PARAMETERS,
{{"rocksdb.num_levels", "11"}}},
+ // Wrong rocksdb.num_levels (non-digital character).
{APP_NAME + "_11",
4,
3,
@@ -452,6 +479,7 @@ TEST_F(meta_app_operation_test, create_app)
app_status::AS_INVALID,
ERR_INVALID_PARAMETERS,
{{"rocksdb.num_levels", "5i"}}},
+ // Create app with rocksdb.num_levels (= 5) succeed.
{APP_NAME + "_11",
4,
3,
@@ -462,6 +490,7 @@ TEST_F(meta_app_operation_test, create_app)
app_status::AS_INVALID,
ERR_OK,
{{"rocksdb.num_levels", "5"}}},
+ // Wrong rocksdb.write_buffer_size (< (16<<20)).
{APP_NAME,
4,
3,
@@ -472,6 +501,7 @@ TEST_F(meta_app_operation_test, create_app)
app_status::AS_INVALID,
ERR_INVALID_PARAMETERS,
{{"rocksdb.write_buffer_size", "1000"}}},
+ // Wrong rocksdb.write_buffer_size (> (512<<20)).
{APP_NAME,
4,
3,
@@ -482,6 +512,7 @@ TEST_F(meta_app_operation_test, create_app)
app_status::AS_INVALID,
ERR_INVALID_PARAMETERS,
{{"rocksdb.write_buffer_size", "1073741824"}}},
+ // Wrong rocksdb.write_buffer_size (non-digital character).
{APP_NAME,
4,
3,
@@ -492,6 +523,7 @@ TEST_F(meta_app_operation_test, create_app)
app_status::AS_INVALID,
ERR_INVALID_PARAMETERS,
{{"rocksdb.write_buffer_size", "n33554432"}}},
+ // Create app with rocksdb.write_buffer_size (= (32<<20)) succeed.
{APP_NAME + "_12",
4,
3,
@@ -502,33 +534,160 @@ TEST_F(meta_app_operation_test, create_app)
app_status::AS_INVALID,
ERR_OK,
{{"rocksdb.write_buffer_size", "33554432"}}},
+ // Process the first request of creating follower app for duplication
from the
+ // source cluster.
+ {DUP_FOLLOWER_APP_NAME,
+ 4,
+ 3,
+ 2,
+ 3,
+ 3,
+ false,
+ app_status::AS_INVALID,
+ ERR_OK,
+ {{duplication_constants::kEnvMasterClusterKey, "source_cluster"},
+ {duplication_constants::kEnvMasterMetasKey, "10.1.2.3:34601"},
+ {duplication_constants::kEnvMasterAppNameKey, DUP_MASTER_APP_NAME},
+ {duplication_constants::kEnvFollowerAppStatusKey,
+ duplication_constants::kEnvFollowerAppStatusCreating}}},
+ // Process the request of creating follower app for duplication from
the wrong
+ // source cluster.
+ {DUP_FOLLOWER_APP_NAME,
+ 4,
+ 3,
+ 2,
+ 3,
+ 3,
+ false,
+ app_status::AS_AVAILABLE,
+ ERR_APP_EXIST,
+ {{duplication_constants::kEnvMasterClusterKey,
"another_source_cluster"},
+ {duplication_constants::kEnvMasterMetasKey, "10.1.2.3:34601"},
+ {duplication_constants::kEnvMasterAppNameKey, DUP_MASTER_APP_NAME},
+ {duplication_constants::kEnvFollowerAppStatusKey,
+ duplication_constants::kEnvFollowerAppStatusCreating}}},
+ // Process the request of creating follower app for duplication
without the env of
+ // creating status.
+ {DUP_FOLLOWER_APP_NAME,
+ 4,
+ 3,
+ 2,
+ 3,
+ 3,
+ false,
+ app_status::AS_AVAILABLE,
+ ERR_APP_EXIST,
+ {
+ {duplication_constants::kEnvMasterClusterKey, "source_cluster"},
+ {duplication_constants::kEnvMasterMetasKey, "10.1.2.3:34601"},
+ {duplication_constants::kEnvMasterAppNameKey,
DUP_MASTER_APP_NAME},
+ }},
+ // Process the repeated request of creating follower app for
duplication from the
+ // source cluster.
+ {DUP_FOLLOWER_APP_NAME,
+ 4,
+ 3,
+ 2,
+ 3,
+ 3,
+ false,
+ app_status::AS_AVAILABLE,
+ ERR_OK,
+ {{duplication_constants::kEnvMasterClusterKey, "source_cluster"},
+ {duplication_constants::kEnvMasterMetasKey, "10.1.2.3:34601"},
+ {duplication_constants::kEnvMasterAppNameKey, DUP_MASTER_APP_NAME},
+ {duplication_constants::kEnvFollowerAppStatusKey,
+ duplication_constants::kEnvFollowerAppStatusCreating}}},
+ // Process the request of marking follower app as created for
duplication from the
+ // source cluster.
+ {DUP_FOLLOWER_APP_NAME,
+ 4,
+ 3,
+ 2,
+ 3,
+ 3,
+ false,
+ app_status::AS_AVAILABLE,
+ ERR_OK,
+ {{duplication_constants::kEnvMasterClusterKey, "source_cluster"},
+ {duplication_constants::kEnvMasterMetasKey, "10.1.2.3:34601"},
+ {duplication_constants::kEnvMasterAppNameKey, DUP_MASTER_APP_NAME},
+ {duplication_constants::kEnvFollowerAppStatusKey,
+ duplication_constants::kEnvFollowerAppStatusCreated}}},
+ // Process the request of creating follower app for duplication from
the source
+ // cluster while it has been marked as created.
+ {DUP_FOLLOWER_APP_NAME,
+ 4,
+ 3,
+ 2,
+ 3,
+ 3,
+ false,
+ app_status::AS_AVAILABLE,
+ ERR_APP_EXIST,
+ {{duplication_constants::kEnvMasterClusterKey, "source_cluster"},
+ {duplication_constants::kEnvMasterMetasKey, "10.1.2.3:34601"},
+ {duplication_constants::kEnvMasterAppNameKey, DUP_MASTER_APP_NAME},
+ {duplication_constants::kEnvFollowerAppStatusKey,
+ duplication_constants::kEnvFollowerAppStatusCreating}}},
+ // Process the repeated request of marking follower app as created for
duplication
+ // from the source cluster.
+ {DUP_FOLLOWER_APP_NAME,
+ 4,
+ 3,
+ 2,
+ 3,
+ 3,
+ false,
+ app_status::AS_AVAILABLE,
+ ERR_OK,
+ {{duplication_constants::kEnvMasterClusterKey, "source_cluster"},
+ {duplication_constants::kEnvMasterMetasKey, "10.1.2.3:34601"},
+ {duplication_constants::kEnvMasterAppNameKey, DUP_MASTER_APP_NAME},
+ {duplication_constants::kEnvFollowerAppStatusKey,
+ duplication_constants::kEnvFollowerAppStatusCreated}}},
+ // Process the request of creating follower app for duplication with
invalid creating
+ // status.
+ {DUP_FOLLOWER_APP_NAME,
+ 4,
+ 3,
+ 2,
+ 3,
+ 3,
+ false,
+ app_status::AS_AVAILABLE,
+ ERR_INVALID_PARAMETERS,
+ {{duplication_constants::kEnvMasterClusterKey, "source_cluster"},
+ {duplication_constants::kEnvMasterMetasKey, "10.1.2.3:34601"},
+ {duplication_constants::kEnvMasterAppNameKey, DUP_MASTER_APP_NAME},
+ {duplication_constants::kEnvFollowerAppStatusKey,
"invalid_creating_status"}}},
};
clear_nodes();
- // keep the number of all nodes greater than that of alive nodes
+ // Keep the number of all nodes greater than that of alive nodes.
const int total_node_count = 10;
auto nodes = ensure_enough_alive_nodes(total_node_count);
- // the meta function level will become freezed once
+ // The meta function level will become freezed once
// alive_nodes * 100 < total_nodes *
_node_live_percentage_threshold_for_update
// even if alive_nodes >= min_live_node_count_for_unfreeze
set_node_live_percentage_threshold_for_update(0);
- // save original FLAGS_min_live_node_count_for_unfreeze
+ // Save original FLAGS_min_live_node_count_for_unfreeze
auto reserved_min_live_node_count_for_unfreeze =
FLAGS_min_live_node_count_for_unfreeze;
- // save original FLAGS_max_allowed_replica_count
+ // Save original FLAGS_max_allowed_replica_count
auto reserved_max_allowed_replica_count = FLAGS_max_allowed_replica_count;
- // keep FLAGS_max_allowed_replica_count fixed in the tests
+ // Keep FLAGS_max_allowed_replica_count fixed in the tests
auto res = update_flag("max_allowed_replica_count", "5");
ASSERT_TRUE(res.is_ok());
- // save original FLAGS_min_allowed_replica_count
+ // Save original FLAGS_min_allowed_replica_count.
auto reserved_min_allowed_replica_count = FLAGS_min_allowed_replica_count;
- for (auto test : tests) {
+ for (const auto &test : tests) {
res = update_flag("min_allowed_replica_count",
std::to_string(test.min_allowed_replica_count));
ASSERT_TRUE(res.is_ok());
@@ -546,12 +705,17 @@ TEST_F(meta_app_operation_test, create_app)
} else if (test.before_status != app_status::AS_INVALID) {
update_app_status(test.before_status);
}
+
auto err = create_app_test(test.partition_count,
test.replica_count,
test.success_if_exist,
test.app_name,
test.envs);
- ASSERT_EQ(err, test.expected_err);
+ ASSERT_EQ(test.expected_err, err);
+
+ if (test.expected_err == ERR_OK) {
+ verify_app_envs(test.app_name, test.envs);
+ }
_ms->set_node_state(nodes, true);
}
@@ -570,53 +734,54 @@ TEST_F(meta_app_operation_test, create_app)
ASSERT_TRUE(all_test_envs.find(option) != all_test_envs.end());
}
}
- // set FLAGS_min_allowed_replica_count successfully
+
+ // Set FLAGS_min_allowed_replica_count successfully.
res = update_flag("min_allowed_replica_count", "2");
ASSERT_TRUE(res.is_ok());
- ASSERT_EQ(FLAGS_min_allowed_replica_count, 2);
+ ASSERT_EQ(2, FLAGS_min_allowed_replica_count);
- // set FLAGS_max_allowed_replica_count successfully
+ // Set FLAGS_max_allowed_replica_count successfully.
res = update_flag("max_allowed_replica_count", "6");
ASSERT_TRUE(res.is_ok());
- ASSERT_EQ(FLAGS_max_allowed_replica_count, 6);
+ ASSERT_EQ(6, FLAGS_max_allowed_replica_count);
- // failed to set FLAGS_min_allowed_replica_count due to individual
validation
+ // Failed to set FLAGS_min_allowed_replica_count due to individual
validation.
res = update_flag("min_allowed_replica_count", "0");
ASSERT_EQ(res.code(), ERR_INVALID_PARAMETERS);
- ASSERT_EQ(FLAGS_min_allowed_replica_count, 2);
+ ASSERT_EQ(2, FLAGS_min_allowed_replica_count);
std::cout << res.description() << std::endl;
- // failed to set FLAGS_max_allowed_replica_count due to individual
validation
+ // Failed to set FLAGS_max_allowed_replica_count due to individual
validation.
res = update_flag("max_allowed_replica_count", "0");
ASSERT_EQ(res.code(), ERR_INVALID_PARAMETERS);
- ASSERT_EQ(FLAGS_max_allowed_replica_count, 6);
+ ASSERT_EQ(6, FLAGS_max_allowed_replica_count);
std::cout << res.description() << std::endl;
- // failed to set FLAGS_min_allowed_replica_count due to grouped validation
+ // Failed to set FLAGS_min_allowed_replica_count due to grouped validation.
res = update_flag("min_allowed_replica_count", "7");
ASSERT_EQ(res.code(), ERR_INVALID_PARAMETERS);
- ASSERT_EQ(FLAGS_min_allowed_replica_count, 2);
+ ASSERT_EQ(2, FLAGS_min_allowed_replica_count);
std::cout << res.description() << std::endl;
- // failed to set FLAGS_max_allowed_replica_count due to grouped validation
+ // Failed to set FLAGS_max_allowed_replica_count due to grouped validation.
res = update_flag("max_allowed_replica_count", "1");
ASSERT_EQ(res.code(), ERR_INVALID_PARAMETERS);
- ASSERT_EQ(FLAGS_max_allowed_replica_count, 6);
+ ASSERT_EQ(6, FLAGS_max_allowed_replica_count);
std::cout << res.description() << std::endl;
- // recover original FLAGS_min_allowed_replica_count
+ // Recover original FLAGS_min_allowed_replica_count.
res = update_flag("min_allowed_replica_count",
std::to_string(reserved_min_allowed_replica_count));
ASSERT_TRUE(res.is_ok());
ASSERT_EQ(FLAGS_min_allowed_replica_count,
reserved_min_allowed_replica_count);
- // recover original FLAGS_max_allowed_replica_count
+ // Recover original FLAGS_max_allowed_replica_count.
res = update_flag("max_allowed_replica_count",
std::to_string(reserved_max_allowed_replica_count));
ASSERT_TRUE(res.is_ok());
- ASSERT_EQ(FLAGS_max_allowed_replica_count,
reserved_max_allowed_replica_count);
+ ASSERT_EQ(reserved_max_allowed_replica_count,
FLAGS_max_allowed_replica_count);
- // recover original FLAGS_min_live_node_count_for_unfreeze
+ // Recover original FLAGS_min_live_node_count_for_unfreeze.
set_min_live_node_count_for_unfreeze(reserved_min_live_node_count_for_unfreeze);
}
diff --git a/src/meta/test/meta_duplication_service_test.cpp
b/src/meta/test/meta_duplication_service_test.cpp
index af312fa9e..a8a6668f9 100644
--- a/src/meta/test/meta_duplication_service_test.cpp
+++ b/src/meta/test/meta_duplication_service_test.cpp
@@ -76,7 +76,7 @@ public:
static const std::string kTestRemoteAppName;
static const int32_t kTestRemoteReplicaCount;
- meta_duplication_service_test() {}
+ meta_duplication_service_test() = default;
duplication_add_response create_dup(const std::string &app_name,
const std::string &remote_cluster,
@@ -198,12 +198,14 @@ public:
dup_svc().create_follower_app_for_duplication(dup, app);
}
- void check_follower_app_if_create_completed(const
std::shared_ptr<duplication_info> &dup)
+ void mark_follower_app_created_for_duplication(const
std::shared_ptr<duplication_info> &dup,
+ const
std::shared_ptr<app_state> &app)
{
- dup_svc().check_follower_app_if_create_completed(dup);
+ dup_svc().mark_follower_app_created_for_duplication(dup, app);
}
- duplication_status::type next_status(const
std::shared_ptr<duplication_info> &dup) const
+ [[nodiscard]] static duplication_status::type
+ next_status(const std::shared_ptr<duplication_info> &dup)
{
return dup->_next_status;
}
@@ -965,108 +967,123 @@ TEST_F(meta_duplication_service_test,
create_follower_app_for_duplication)
}
}
-TEST_F(meta_duplication_service_test, check_follower_app_if_create_completed)
+TEST_F(meta_duplication_service_test,
mark_follower_app_created_for_duplication)
{
struct test_case
{
- int32_t remote_replica_count;
std::vector<std::string> fail_cfg_name;
std::vector<std::string> fail_cfg_action;
- bool is_altering;
+ int32_t remote_replica_count;
duplication_status::type cur_status;
duplication_status::type next_status;
+ bool is_altering;
} test_cases[] = {// 3 remote replicas with both primary and secondaries
valid.
- {3,
- {"create_app_ok"},
- {"void(true,2,0)"},
- false,
+ {{"on_follower_app_created", "create_app_ok"},
+ {"void(ERR_OK)", "void(true,2,0)"},
+ 3,
duplication_status::DS_LOG,
- duplication_status::DS_INIT},
+ duplication_status::DS_INIT,
+ false},
+ // The follower app failed to be marked as created.
+ {{"on_follower_app_created", "create_app_ok"},
+ {"void(ERR_TIMEOUT)", "void(true,2,0)"},
+ 3,
+ duplication_status::DS_APP,
+ duplication_status::DS_INIT,
+ false},
+ //
// 3 remote replicas with primary invalid and all
secondaries valid.
- {3,
- {"create_app_ok"},
- {"void(false,2,0)"},
- false,
+ {{"on_follower_app_created", "create_app_ok"},
+ {"void(ERR_OK)", "void(false,2,0)"},
+ 3,
duplication_status::DS_APP,
- duplication_status::DS_INIT},
+ duplication_status::DS_INIT,
+ false},
// 3 remote replicas with primary valid and only one
secondary present
// and valid.
- {3,
- {"create_app_ok"},
- {"void(true,1,0)"},
- false,
+ {{"on_follower_app_created", "create_app_ok"},
+ {"void(ERR_OK)", "void(true,1,0)"},
+ 3,
duplication_status::DS_LOG,
- duplication_status::DS_INIT},
+ duplication_status::DS_INIT,
+ false},
// 3 remote replicas with primary valid and one
secondary invalid.
- {3,
- {"create_app_ok"},
- {"void(true,1,1)"},
- false,
+ {{"on_follower_app_created", "create_app_ok"},
+ {"void(ERR_OK)", "void(true,1,1)"},
+ 3,
duplication_status::DS_APP,
- duplication_status::DS_INIT},
+ duplication_status::DS_INIT,
+ false},
// 3 remote replicas with primary valid and only one
secondary present
// and invalid.
- {3,
- {"create_app_ok"},
- {"void(true,0,1)"},
- false,
+ {{"on_follower_app_created", "create_app_ok"},
+ {"void(ERR_OK)", "void(true,0,1)"},
+ 3,
duplication_status::DS_APP,
- duplication_status::DS_INIT},
+ duplication_status::DS_INIT,
+ false},
// 3 remote replicas with primary valid and both
secondaries absent.
- {3,
- {"create_app_ok"},
- {"void(true,0,0)"},
- false,
- duplication_status::DS_APP,
- duplication_status::DS_INIT},
+ {
+ {"on_follower_app_created", "create_app_ok"},
+ {"void(ERR_OK)", "void(true,0,0)"},
+ 3,
+ duplication_status::DS_APP,
+ duplication_status::DS_INIT,
+ false,
+ },
// 1 remote replicas with primary valid.
- {1,
- {"create_app_ok"},
- {"void(true,0,0)"},
- false,
- duplication_status::DS_LOG,
- duplication_status::DS_INIT},
+ {
+ {"on_follower_app_created", "create_app_ok"},
+ {"void(ERR_OK)", "void(true,0,0)"},
+ 1,
+ duplication_status::DS_LOG,
+ duplication_status::DS_INIT,
+ false,
+ },
// 1 remote replicas with primary invalid.
- {1,
- {"create_app_ok"},
- {"void(false,0,0)"},
- false,
- duplication_status::DS_APP,
- duplication_status::DS_INIT},
+ {
+ {"on_follower_app_created", "create_app_ok"},
+ {"void(ERR_OK)", "void(false,0,0)"},
+ 1,
+ duplication_status::DS_APP,
+ duplication_status::DS_INIT,
+ false,
+ },
// The case is just a "palace holder", actually
// `check_follower_app_if_create_completed` would fail
by default
// in unit test.
- {3,
- {"create_app_failed"},
- {"off()"},
- false,
+ {{"on_follower_app_created", "create_app_failed"},
+ {"void(ERR_OK)", "off()"},
+ 3,
duplication_status::DS_APP,
- duplication_status::DS_INIT},
- {3,
- {"create_app_ok", "persist_dup_status_failed"},
- {"void(true,2,0)", "return()"},
- true,
+ duplication_status::DS_INIT,
+ false},
+ {{"on_follower_app_created", "create_app_ok",
"persist_dup_status_failed"},
+ {"void(ERR_OK)", "void(true,2,0)", "return()"},
+ 3,
duplication_status::DS_APP,
- duplication_status::DS_LOG}};
+ duplication_status::DS_LOG,
+ true}};
size_t i = 0;
for (const auto &test : test_cases) {
- const auto &app_name =
fmt::format("check_follower_app_if_create_completed_test_{}", i++);
+ const auto &app_name =
+ fmt::format("mark_follower_app_created_for_duplication_test_{}",
i++);
create_app(app_name);
auto app = find_app(app_name);
auto dup_add_resp = create_dup(app_name, test.remote_replica_count);
auto dup = app->duplications[dup_add_resp.dupid];
- // 'check_follower_app_if_create_completed' must execute under
duplication_status::DS_APP,
- // so force update it.
+ // 'mark_follower_app_created_for_duplication' must execute under
+ // duplication_status::DS_APP, so force update it.
force_update_dup_status(dup, duplication_status::DS_APP);
fail::setup();
for (int i = 0; i < test.fail_cfg_name.size(); i++) {
fail::cfg(test.fail_cfg_name[i], test.fail_cfg_action[i]);
}
- check_follower_app_if_create_completed(dup);
+ mark_follower_app_created_for_duplication(dup, app);
wait_all();
fail::teardown();
diff --git a/src/meta/test/meta_state/meta_state_service.cpp
b/src/meta/test/meta_state/meta_state_service.cpp
index 7c1db0662..6ca22721d 100644
--- a/src/meta/test/meta_state/meta_state_service.cpp
+++ b/src/meta/test/meta_state/meta_state_service.cpp
@@ -99,7 +99,7 @@ void provider_basic_test(const service_creator_func
&service_creator,
service->delete_node("/1", false,
META_STATE_SERVICE_SIMPLE_TEST_CALLBACK, expect_ok)
->wait();
}
- // set & get data
+ // create & get data
{
dsn::binary_writer writer;
writer.write(0xdeadbeef);
@@ -118,7 +118,10 @@ void provider_basic_test(const service_creator_func
&service_creator,
CHECK_EQ(0xdeadbeef, read_value);
})
->wait();
- writer = dsn::binary_writer();
+ }
+ // set & get data
+ {
+ dsn::binary_writer writer;
writer.write(0xbeefdead);
service
->set_data(
diff --git a/src/replica/duplication/replica_follower.cpp
b/src/replica/duplication/replica_follower.cpp
index 057358f59..9aa5628b1 100644
--- a/src/replica/duplication/replica_follower.cpp
+++ b/src/replica/duplication/replica_follower.cpp
@@ -60,8 +60,8 @@ void replica_follower::init_master_info()
{
const auto &envs = _replica->get_app_info()->envs;
- const auto &cluster_name =
envs.find(duplication_constants::kDuplicationEnvMasterClusterKey);
- const auto &metas =
envs.find(duplication_constants::kDuplicationEnvMasterMetasKey);
+ const auto &cluster_name =
envs.find(duplication_constants::kEnvMasterClusterKey);
+ const auto &metas = envs.find(duplication_constants::kEnvMasterMetasKey);
if (cluster_name == envs.end() || metas == envs.end()) {
return;
}
@@ -70,7 +70,7 @@ void replica_follower::init_master_info()
_master_cluster_name = cluster_name->second;
- const auto &app_name =
envs.find(duplication_constants::kDuplicationEnvMasterAppNameKey);
+ const auto &app_name =
envs.find(duplication_constants::kEnvMasterAppNameKey);
if (app_name == envs.end()) {
// The version of meta server of master cluster is old(< v2.6.0), thus
the app name of
// the follower cluster is the same with master cluster.
diff --git a/src/replica/duplication/test/replica_follower_test.cpp
b/src/replica/duplication/test/replica_follower_test.cpp
index 45cf7e9df..c47c90a31 100644
--- a/src/replica/duplication/test/replica_follower_test.cpp
+++ b/src/replica/duplication/test/replica_follower_test.cpp
@@ -62,9 +62,8 @@ public:
void update_mock_replica(const dsn::app_info &app)
{
bool is_duplication_follower =
-
(app.envs.find(duplication_constants::kDuplicationEnvMasterClusterKey) !=
- app.envs.end()) &&
-
(app.envs.find(duplication_constants::kDuplicationEnvMasterMetasKey) !=
app.envs.end());
+ (app.envs.find(duplication_constants::kEnvMasterClusterKey) !=
app.envs.end()) &&
+ (app.envs.find(duplication_constants::kEnvMasterMetasKey) !=
app.envs.end());
_mock_replica = stub->generate_replica_ptr(
app, gpid(2, 1), partition_status::PS_PRIMARY, 1, false,
is_duplication_follower);
}
@@ -115,9 +114,8 @@ public:
void test_init_master_info(const std::string &expected_master_app_name)
{
-
_app_info.envs.emplace(duplication_constants::kDuplicationEnvMasterClusterKey,
- kTestMasterClusterName);
-
_app_info.envs.emplace(duplication_constants::kDuplicationEnvMasterMetasKey,
+ _app_info.envs.emplace(duplication_constants::kEnvMasterClusterKey,
kTestMasterClusterName);
+ _app_info.envs.emplace(duplication_constants::kEnvMasterMetasKey,
"127.0.0.1:34801,127.0.0.1:34802,127.0.0.1:34803");
update_mock_replica(_app_info);
@@ -156,16 +154,14 @@ TEST_P(replica_follower_test,
test_init_master_info_without_master_app_env)
TEST_P(replica_follower_test, test_init_master_info_with_master_app_env)
{
static const std::string kTestAnotherMasterAppName("another_follower");
-
_app_info.envs.emplace(duplication_constants::kDuplicationEnvMasterAppNameKey,
- kTestAnotherMasterAppName);
+ _app_info.envs.emplace(duplication_constants::kEnvMasterAppNameKey,
kTestAnotherMasterAppName);
test_init_master_info(kTestAnotherMasterAppName);
}
TEST_P(replica_follower_test, test_duplicate_checkpoint)
{
-
_app_info.envs.emplace(duplication_constants::kDuplicationEnvMasterClusterKey,
- kTestMasterClusterName);
-
_app_info.envs.emplace(duplication_constants::kDuplicationEnvMasterMetasKey,
+ _app_info.envs.emplace(duplication_constants::kEnvMasterClusterKey,
kTestMasterClusterName);
+ _app_info.envs.emplace(duplication_constants::kEnvMasterMetasKey,
"127.0.0.1:34801,127.0.0.1:34802,127.0.0.1:34803");
update_mock_replica(_app_info);
@@ -184,9 +180,8 @@ TEST_P(replica_follower_test, test_duplicate_checkpoint)
TEST_P(replica_follower_test,
test_async_duplicate_checkpoint_from_master_replica)
{
-
_app_info.envs.emplace(duplication_constants::kDuplicationEnvMasterClusterKey,
- kTestMasterClusterName);
-
_app_info.envs.emplace(duplication_constants::kDuplicationEnvMasterMetasKey,
+ _app_info.envs.emplace(duplication_constants::kEnvMasterClusterKey,
kTestMasterClusterName);
+ _app_info.envs.emplace(duplication_constants::kEnvMasterMetasKey,
"127.0.0.1:34801,127.0.0.1:34802,127.0.0.1:34803");
update_mock_replica(_app_info);
@@ -207,12 +202,11 @@ TEST_P(replica_follower_test,
test_async_duplicate_checkpoint_from_master_replic
TEST_P(replica_follower_test, test_update_master_replica_config)
{
-
_app_info.envs.emplace(duplication_constants::kDuplicationEnvMasterClusterKey,
- kTestMasterClusterName);
-
_app_info.envs.emplace(duplication_constants::kDuplicationEnvMasterMetasKey,
+ _app_info.envs.emplace(duplication_constants::kEnvMasterClusterKey,
kTestMasterClusterName);
+ _app_info.envs.emplace(duplication_constants::kEnvMasterMetasKey,
"127.0.0.1:34801,127.0.0.1:34802,127.0.0.1:34803");
update_mock_replica(_app_info);
- auto follower = _mock_replica->get_replica_follower();
+ auto *follower = _mock_replica->get_replica_follower();
query_cfg_response resp;
ASSERT_EQ(update_master_replica_config(follower, resp),
ERR_INCONSISTENT_STATE);
@@ -265,9 +259,8 @@ TEST_P(replica_follower_test,
test_update_master_replica_config)
TEST_P(replica_follower_test, test_nfs_copy_checkpoint)
{
-
_app_info.envs.emplace(duplication_constants::kDuplicationEnvMasterClusterKey,
- kTestMasterClusterName);
-
_app_info.envs.emplace(duplication_constants::kDuplicationEnvMasterMetasKey,
+ _app_info.envs.emplace(duplication_constants::kEnvMasterClusterKey,
kTestMasterClusterName);
+ _app_info.envs.emplace(duplication_constants::kEnvMasterMetasKey,
"127.0.0.1:34801,127.0.0.1:34802,127.0.0.1:34803");
update_mock_replica(_app_info);
init_nfs();
diff --git a/src/replica/replica_stub.cpp b/src/replica/replica_stub.cpp
index de2b78973..6389ed50c 100644
--- a/src/replica/replica_stub.cpp
+++ b/src/replica/replica_stub.cpp
@@ -1844,10 +1844,8 @@ void replica_stub::open_replica(
bool is_duplication_follower =
((configuration_update != nullptr) &&
(configuration_update->type == config_type::CT_ASSIGN_PRIMARY) &&
-
(app.envs.find(duplication_constants::kDuplicationEnvMasterClusterKey) !=
- app.envs.end()) &&
-
(app.envs.find(duplication_constants::kDuplicationEnvMasterMetasKey) !=
- app.envs.end()));
+ (app.envs.find(duplication_constants::kEnvMasterClusterKey) !=
app.envs.end()) &&
+ (app.envs.find(duplication_constants::kEnvMasterMetasKey) !=
app.envs.end()));
// NOTICE: when we don't need execute restore-process, we should
remove a.b.pegasus
// directory because it don't contain the valid data dir and also we
need create a new
diff --git a/src/rpc/rpc_stream.h b/src/rpc/rpc_stream.h
index f9953e8d8..ffaa049e2 100644
--- a/src/rpc/rpc_stream.h
+++ b/src/rpc/rpc_stream.h
@@ -83,7 +83,9 @@ typedef ::dsn::ref_ptr<rpc_read_stream> rpc_read_stream_ptr;
class rpc_write_stream : public binary_writer
{
public:
- rpc_write_stream(message_ex *msg)
+ rpc_write_stream() = delete;
+
+ explicit rpc_write_stream(message_ex *msg)
: _msg(msg), _last_write_next_committed(true),
_last_write_next_total_size(0)
{
}
@@ -104,16 +106,16 @@ public:
}
}
- virtual ~rpc_write_stream() { flush(); }
-
- virtual void flush() override
+ ~rpc_write_stream() override
{
- binary_writer::flush();
- commit_buffer();
+ // Avoid calling virtual functions in destructor.
+ flush_internal();
}
+ void flush() override { flush_internal(); }
+
private:
- virtual void create_new_buffer(size_t size, /*out*/ blob &bb) override
+ void create_new_buffer(size_t size, /*out*/ blob &bb) override
{
commit_buffer();
@@ -127,10 +129,20 @@ private:
_last_write_next_committed = false;
}
-private:
+ void flush_internal()
+ {
+ binary_writer::flush();
+ commit_buffer();
+ }
+
message_ex *_msg;
bool _last_write_next_committed;
int _last_write_next_total_size;
+
+ DISALLOW_COPY_AND_ASSIGN(rpc_write_stream);
+ DISALLOW_MOVE_AND_ASSIGN(rpc_write_stream);
};
-typedef ::dsn::ref_ptr<rpc_write_stream> rpc_write_stream_ptr;
+
+using rpc_write_stream_ptr = dsn::ref_ptr<rpc_write_stream>;
+
} // namespace dsn
diff --git a/src/utils/binary_writer.cpp b/src/utils/binary_writer.cpp
index 9234ce914..64049dfe6 100644
--- a/src/utils/binary_writer.cpp
+++ b/src/utils/binary_writer.cpp
@@ -32,33 +32,32 @@
#include "utils/blob.h"
namespace dsn {
-int binary_writer::_reserved_size_per_buffer_static = 256;
-binary_writer::binary_writer(int reserveBufferSize)
+const int binary_writer::kReservedSizePerBuffer = 256;
+
+binary_writer::binary_writer() : binary_writer(0) {}
+
+binary_writer::binary_writer(int reserved_buffer_size)
+ : _current_buffer(nullptr),
+ _current_offset(0),
+ _current_buffer_length(0),
+ _total_size(0),
+ _reserved_size_per_buffer((reserved_buffer_size == 0) ?
kReservedSizePerBuffer
+ :
reserved_buffer_size)
{
- _total_size = 0;
_buffers.reserve(1);
- _reserved_size_per_buffer = (reserveBufferSize == 0) ?
_reserved_size_per_buffer_static
- : reserveBufferSize;
- _current_buffer = nullptr;
- _current_offset = 0;
- _current_buffer_length = 0;
}
binary_writer::binary_writer(blob &buffer)
+ : _buffers({buffer}),
+ _current_buffer(const_cast<char *>(buffer.data())),
+ _current_offset(0),
+ _current_buffer_length(static_cast<int>(buffer.length())),
+ _total_size(0),
+ _reserved_size_per_buffer(kReservedSizePerBuffer)
{
- _total_size = 0;
- _buffers.reserve(1);
- _reserved_size_per_buffer = _reserved_size_per_buffer_static;
-
- _buffers.push_back(buffer);
- _current_buffer = (char *)buffer.data();
- _current_offset = 0;
- _current_buffer_length = buffer.length();
}
-binary_writer::~binary_writer() {}
-
void binary_writer::flush() { commit(); }
void binary_writer::create_buffer(size_t size)
@@ -200,4 +199,5 @@ bool binary_writer::backup(int count)
_total_size -= count;
return true;
}
+
} // namespace dsn
diff --git a/src/utils/binary_writer.h b/src/utils/binary_writer.h
index 2640567f0..7bce6139d 100644
--- a/src/utils/binary_writer.h
+++ b/src/utils/binary_writer.h
@@ -34,15 +34,17 @@
#include <vector>
#include "blob.h"
+#include "utils/ports.h"
namespace dsn {
class binary_writer
{
public:
- binary_writer(int reserved_buffer_size = 0);
- binary_writer(blob &buffer);
- virtual ~binary_writer();
+ binary_writer();
+ explicit binary_writer(int reserved_buffer_size);
+ explicit binary_writer(blob &buffer);
+ virtual ~binary_writer() = default;
virtual void flush();
@@ -95,7 +97,10 @@ private:
int _total_size;
int _reserved_size_per_buffer;
- static int _reserved_size_per_buffer_static;
+ static const int kReservedSizePerBuffer;
+
+ DISALLOW_COPY_AND_ASSIGN(binary_writer);
+ DISALLOW_MOVE_AND_ASSIGN(binary_writer);
};
//--------------- inline implementation -------------------
diff --git a/src/utils/fmt_utils.h b/src/utils/fmt_utils.h
index 9624bb881..2d18201c1 100644
--- a/src/utils/fmt_utils.h
+++ b/src/utils/fmt_utils.h
@@ -26,4 +26,4 @@
}
#define USER_DEFINED_ENUM_FORMATTER(type)
\
- inline auto format_as(type e)->int { return e; }
+ inline int format_as(type e) { return e; }
diff --git a/src/utils/ports.h b/src/utils/ports.h
index 718810f05..7e76e2c4a 100644
--- a/src/utils/ports.h
+++ b/src/utils/ports.h
@@ -63,9 +63,21 @@
#define dsn_likely(pred) (__builtin_expect((pred), 1))
#define dsn_unlikely(pred) (__builtin_expect((pred), 0))
-#define DISALLOW_COPY_AND_ASSIGN(TypeName)
\
- TypeName(const TypeName &) = delete;
\
- void operator=(const TypeName &) = delete
+#define DECLARE_COPY_AND_ASSIGN(type, action)
\
+ type(const type &) = action;
\
+ type &operator=(const type &) = action
+
+#define DECLARE_MOVE_AND_ASSIGN(type, action)
\
+ type(type &&) = action;
\
+ type &operator=(type &&) = action
+
+#define DEFAULT_COPY_AND_ASSIGN(type) DECLARE_COPY_AND_ASSIGN(type, default)
+
+#define DEFAULT_MOVE_AND_ASSIGN(type) DECLARE_MOVE_AND_ASSIGN(type, default)
+
+#define DISALLOW_COPY_AND_ASSIGN(type) DECLARE_COPY_AND_ASSIGN(type, delete)
+
+#define DISALLOW_MOVE_AND_ASSIGN(type) DECLARE_MOVE_AND_ASSIGN(type, delete)
#if defined OS_LINUX || defined OS_CYGWIN
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@pegasus.apache.org
For additional commands, e-mail: commits-h...@pegasus.apache.org
