diff --git a/apps/emqx/test/emqx_common_test_helpers.erl b/apps/emqx/test/emqx_common_test_helpers.erl
index 18919103c..6f2b05161 100644
--- a/apps/emqx/test/emqx_common_test_helpers.erl
+++ b/apps/emqx/test/emqx_common_test_helpers.erl
@@ -70,8 +70,8 @@
emqx_cluster/2,
start_ekka/0,
start_epmd/0,
- start_slave/2,
- stop_slave/1,
+ start_peer/2,
+ stop_peer/1,
listener_port/2
]).
@@ -734,13 +734,11 @@ emqx_cluster(Specs0, CommonOpts) ->
%% Lower level starting API
--spec start_slave(shortname(), node_opts()) -> nodename().
-start_slave(Name, Opts) when is_list(Opts) ->
- start_slave(Name, maps:from_list(Opts));
-start_slave(Name, Opts) when is_map(Opts) ->
- SlaveMod = maps:get(peer_mod, Opts, ct_slave),
+-spec start_peer(shortname(), node_opts()) -> nodename().
+start_peer(Name, Opts) when is_list(Opts) ->
+ start_peer(Name, maps:from_list(Opts));
+start_peer(Name, Opts) when is_map(Opts) ->
Node = node_name(Name),
- put_peer_mod(Node, SlaveMod),
Cookie = atom_to_list(erlang:get_cookie()),
PrivDataDir = maps:get(priv_data_dir, Opts, "/tmp"),
NodeDataDir = filename:join([
@@ -750,19 +748,13 @@ start_slave(Name, Opts) when is_map(Opts) ->
]),
DoStart =
fun() ->
- case SlaveMod of
- ct_slave ->
- ct:pal("~p: node data dir: ~s", [Node, NodeDataDir]),
- Envs = [
- {"HOCON_ENV_OVERRIDE_PREFIX", "EMQX_"},
- {"EMQX_NODE__COOKIE", Cookie},
- {"EMQX_NODE__DATA_DIR", NodeDataDir}
- ],
- emqx_cth_peer:start(Node, erl_flags(), Envs);
- slave ->
- Envs = [{"HOCON_ENV_OVERRIDE_PREFIX", "EMQX_"}],
- emqx_cth_peer:start(Node, ebin_path(), Envs)
- end
+ ct:pal("~p: node data dir: ~s", [Node, NodeDataDir]),
+ Envs = [
+ {"HOCON_ENV_OVERRIDE_PREFIX", "EMQX_"},
+ {"EMQX_NODE__COOKIE", Cookie},
+ {"EMQX_NODE__DATA_DIR", NodeDataDir}
+ ],
+ emqx_cth_peer:start(Node, erl_flags(), Envs)
end,
case DoStart() of
{ok, _} ->
@@ -778,7 +770,7 @@ start_slave(Name, Opts) when is_map(Opts) ->
Node.
%% Node stopping
-stop_slave(Node0) ->
+stop_peer(Node0) ->
Node = node_name(Node0),
emqx_cth_peer:stop(Node).
@@ -939,7 +931,7 @@ setup_node(Node, Opts) when is_map(Opts) ->
ignore ->
ok;
Err ->
- stop_slave(Node),
+ stop_peer(Node),
error({failed_to_join_cluster, #{node => Node, error => Err}})
end
end,
@@ -956,19 +948,6 @@ set_env_once(Var, Value) ->
end,
ok.
-put_peer_mod(Node, SlaveMod) ->
- put({?MODULE, Node}, SlaveMod),
- ok.
-
-get_peer_mod(Node) ->
- case get({?MODULE, Node}) of
- undefined -> ct_slave;
- SlaveMod -> SlaveMod
- end.
-
-erase_peer_mod(Node) ->
- erase({?MODULE, Node}).
-
node_name(Name) ->
case string:tokens(atom_to_list(Name), "@") of
[_Name, _Host] ->
diff --git a/apps/emqx/test/emqx_cth_cluster.erl b/apps/emqx/test/emqx_cth_cluster.erl
index 029907f57..d53bb1e90 100644
--- a/apps/emqx/test/emqx_cth_cluster.erl
+++ b/apps/emqx/test/emqx_cth_cluster.erl
@@ -52,6 +52,7 @@
-define(TIMEOUT_NODE_START_MS, 15000).
-define(TIMEOUT_APPS_START_MS, 30000).
-define(TIMEOUT_NODE_STOP_S, 15).
+-define(TIMEOUT_CLUSTER_WAIT_MS, timer:seconds(10)).
%%
@@ -91,11 +92,7 @@
%% Working directory
%% If this directory is not empty, starting up the node applications will fail
%% Default: "${ClusterOpts.work_dir}/${nodename}"
- work_dir => file:name(),
-
- % Tooling to manage nodes
- % Default: `ct_slave`.
- driver => ct_slave | slave
+ work_dir => file:name()
}}.
-spec start([nodespec()], ClusterOpts) ->
@@ -118,11 +115,52 @@ start(NodeSpecs) ->
% 2. Start applications needed to enable clustering
% Generally, this causes some applications to restart, but we deliberately don't
% start them yet.
- _ = lists:foreach(fun run_node_phase_cluster/1, NodeSpecs),
+ ShouldAppearInRunningNodes = lists:map(fun run_node_phase_cluster/1, NodeSpecs),
+ IsClustered = lists:member(true, ShouldAppearInRunningNodes),
% 3. Start applications after cluster is formed
% Cluster-joins are complete, so they shouldn't restart in the background anymore.
_ = emqx_utils:pmap(fun run_node_phase_apps/1, NodeSpecs, ?TIMEOUT_APPS_START_MS),
- [Node || #{name := Node} <- NodeSpecs].
+ Nodes = [Node || #{name := Node} <- NodeSpecs],
+ %% 4. Wait for the nodes to cluster
+ case IsClustered of
+ true ->
+ ok = wait_clustered(Nodes, ?TIMEOUT_CLUSTER_WAIT_MS);
+ false ->
+ ok
+ end,
+ Nodes.
+
+%% Wait until all nodes see all nodes as mria running nodes
+wait_clustered(Nodes, Timeout) ->
+ Check = fun(Node) ->
+ Running = erpc:call(Node, mria, running_nodes, []),
+ case Nodes -- Running of
+ [] ->
+ true;
+ NotRunning ->
+ {false, NotRunning}
+ end
+ end,
+ wait_clustered(Nodes, Check, deadline(Timeout)).
+
+wait_clustered([], _Check, _Deadline) ->
+ ok;
+wait_clustered([Node | Nodes] = All, Check, Deadline) ->
+ IsOverdue = is_overdue(Deadline),
+ case Check(Node) of
+ true ->
+ wait_clustered(Nodes, Check, Deadline);
+ {false, NodesNotRunnging} when IsOverdue ->
+ error(
+ {timeout, #{
+ checking_from_node => Node,
+ nodes_not_running => NodesNotRunnging
+ }}
+ );
+ {false, Nodes} ->
+ timer:sleep(100),
+ wait_clustered(All, Check, Deadline)
+ end.
restart(Node, Spec) ->
ct:pal("Stopping peer node ~p", [Node]),
@@ -162,8 +200,7 @@ mk_init_nodespec(N, Name, NodeOpts, ClusterOpts) ->
role => core,
apps => [],
base_port => BasePort,
- work_dir => filename:join([WorkDir, Node]),
- driver => ct_slave
+ work_dir => filename:join([WorkDir, Node])
},
maps:merge(Defaults, NodeOpts).
@@ -309,15 +346,21 @@ start_bare_nodes(Names, Timeout) ->
end,
Names
),
- Deadline = erlang:monotonic_time() + erlang:convert_time_unit(Timeout, millisecond, nanosecond),
+ Deadline = deadline(Timeout),
Nodes = wait_boot_complete(Waits, Deadline),
lists:foreach(fun(Node) -> pong = net_adm:ping(Node) end, Nodes),
Nodes.
+deadline(Timeout) ->
+ erlang:monotonic_time() + erlang:convert_time_unit(Timeout, millisecond, nanosecond).
+
+is_overdue(Deadline) ->
+ erlang:monotonic_time() > Deadline.
+
wait_boot_complete([], _) ->
[];
wait_boot_complete(Waits, Deadline) ->
- case erlang:monotonic_time() > Deadline of
+ case is_overdue(Deadline) of
true ->
error({timeout, Waits});
false ->
@@ -340,11 +383,11 @@ node_init(Node) ->
ok = snabbkaffe:forward_trace(Node),
ok.
+%% Returns 'true' if this node should appear in running nodes list.
run_node_phase_cluster(Spec = #{name := Node}) ->
ok = load_apps(Node, Spec),
ok = start_apps_clustering(Node, Spec),
- ok = maybe_join_cluster(Node, Spec),
- ok.
+ maybe_join_cluster(Node, Spec).
run_node_phase_apps(Spec = #{name := Node}) ->
ok = start_apps(Node, Spec),
@@ -368,18 +411,20 @@ start_apps(Node, #{apps := Apps} = Spec) ->
suite_opts(Spec) ->
maps:with([work_dir, boot_type], Spec).
+%% Returns 'true' if this node should appear in the cluster.
maybe_join_cluster(_Node, #{boot_type := restart}) ->
%% when restart, the node should already be in the cluster
%% hence no need to (re)join
- ok;
+ true;
maybe_join_cluster(_Node, #{role := replicant}) ->
- ok;
+ true;
maybe_join_cluster(Node, Spec) ->
case get_cluster_seeds(Spec) of
[JoinTo | _] ->
- ok = join_cluster(Node, JoinTo);
+ ok = join_cluster(Node, JoinTo),
+ true;
[] ->
- ok
+ false
end.
join_cluster(Node, JoinTo) ->
diff --git a/apps/emqx/test/emqx_shared_sub_SUITE.erl b/apps/emqx/test/emqx_shared_sub_SUITE.erl
index cc6908fb6..8af3b2190 100644
--- a/apps/emqx/test/emqx_shared_sub_SUITE.erl
+++ b/apps/emqx/test/emqx_shared_sub_SUITE.erl
@@ -63,7 +63,6 @@ init_per_suite(Config) ->
end,
emqx_common_test_helpers:boot_modules(all),
emqx_common_test_helpers:start_apps([]),
- emqx_logger:set_log_level(debug),
[{dist_pid, DistPid} | Config].
end_per_suite(Config) ->
@@ -575,7 +574,7 @@ t_local(Config) when is_list(Config) ->
<<"sticky_group">> => sticky
},
- Node = start_slave('local_shared_sub_local_1', 21999),
+ Node = start_peer('local_shared_sub_local_1', 21999),
ok = ensure_group_config(GroupConfig),
ok = ensure_group_config(Node, GroupConfig),
@@ -606,7 +605,7 @@ t_local(Config) when is_list(Config) ->
emqtt:stop(ConnPid1),
emqtt:stop(ConnPid2),
- stop_slave(Node),
+ stop_peer(Node),
?assertEqual(local, emqx_shared_sub:strategy(<<"local_group">>)),
?assertEqual(local, RemoteLocalGroupStrategy),
@@ -628,7 +627,7 @@ t_remote(Config) when is_list(Config) ->
<<"sticky_group">> => sticky
},
- Node = start_slave('remote_shared_sub_remote_1', 21999),
+ Node = start_peer('remote_shared_sub_remote_1', 21999),
ok = ensure_group_config(GroupConfig),
ok = ensure_group_config(Node, GroupConfig),
@@ -664,7 +663,7 @@ t_remote(Config) when is_list(Config) ->
after
emqtt:stop(ConnPidLocal),
emqtt:stop(ConnPidRemote),
- stop_slave(Node)
+ stop_peer(Node)
end.
t_local_fallback(Config) when is_list(Config) ->
@@ -677,7 +676,7 @@ t_local_fallback(Config) when is_list(Config) ->
Topic = <<"local_foo/bar">>,
ClientId1 = <<"ClientId1">>,
ClientId2 = <<"ClientId2">>,
- Node = start_slave('local_fallback_shared_sub_1', 11888),
+ Node = start_peer('local_fallback_shared_sub_1', 11888),
{ok, ConnPid1} = emqtt:start_link([{clientid, ClientId1}]),
{ok, _} = emqtt:connect(ConnPid1),
@@ -693,7 +692,7 @@ t_local_fallback(Config) when is_list(Config) ->
{true, UsedSubPid2} = last_message(<<"hello2">>, [ConnPid1], 2_000),
emqtt:stop(ConnPid1),
- stop_slave(Node),
+ stop_peer(Node),
?assertEqual(UsedSubPid1, UsedSubPid2),
ok.
@@ -1253,7 +1252,7 @@ recv_msgs(Count, Msgs) ->
Msgs
end.
-start_slave(Name, Port) ->
+start_peer(Name, Port) ->
{ok, Node} = emqx_cth_peer:start_link(
Name,
ebin_path()
@@ -1262,7 +1261,7 @@ start_slave(Name, Port) ->
setup_node(Node, Port),
Node.
-stop_slave(Node) ->
+stop_peer(Node) ->
rpc:call(Node, mria, leave, []),
emqx_cth_peer:stop(Node).
diff --git a/apps/emqx_bridge_gcp_pubsub/test/emqx_bridge_gcp_pubsub_consumer_SUITE.erl b/apps/emqx_bridge_gcp_pubsub/test/emqx_bridge_gcp_pubsub_consumer_SUITE.erl
index 86f81277c..b747e9262 100644
--- a/apps/emqx_bridge_gcp_pubsub/test/emqx_bridge_gcp_pubsub_consumer_SUITE.erl
+++ b/apps/emqx_bridge_gcp_pubsub/test/emqx_bridge_gcp_pubsub_consumer_SUITE.erl
@@ -588,7 +588,6 @@ cluster(Config) ->
[
{apps, [emqx_conf, emqx_rule_engine, emqx_bridge]},
{listener_ports, []},
- {peer_mod, slave},
{priv_data_dir, PrivDataDir},
{load_schema, true},
{start_autocluster, true},
@@ -611,7 +610,7 @@ start_cluster(Cluster) ->
Nodes = lists:map(
fun({Name, Opts}) ->
ct:pal("starting ~p", [Name]),
- emqx_common_test_helpers:start_slave(Name, Opts)
+ emqx_common_test_helpers:start_peer(Name, Opts)
end,
Cluster
),
@@ -620,7 +619,7 @@ start_cluster(Cluster) ->
emqx_utils:pmap(
fun(N) ->
ct:pal("stopping ~p", [N]),
- emqx_common_test_helpers:stop_slave(N)
+ emqx_common_test_helpers:stop_peer(N)
end,
Nodes
)
diff --git a/apps/emqx_bridge_kafka/test/emqx_bridge_kafka_impl_consumer_SUITE.erl b/apps/emqx_bridge_kafka/test/emqx_bridge_kafka_impl_consumer_SUITE.erl
index 4fd08c154..1646ddabe 100644
--- a/apps/emqx_bridge_kafka/test/emqx_bridge_kafka_impl_consumer_SUITE.erl
+++ b/apps/emqx_bridge_kafka/test/emqx_bridge_kafka_impl_consumer_SUITE.erl
@@ -1069,20 +1069,12 @@ setup_and_start_listeners(Node, NodeOpts) ->
cluster(Config) ->
PrivDataDir = ?config(priv_dir, Config),
- PeerModule =
- case os:getenv("IS_CI") of
- false ->
- slave;
- _ ->
- ct_slave
- end,
ExtraEnvHandlerHook = setup_group_subscriber_spy_fn(),
Cluster = emqx_common_test_helpers:emqx_cluster(
[core, core],
[
{apps, [emqx_conf, emqx_rule_engine, emqx_bridge_kafka, emqx_bridge]},
{listener_ports, []},
- {peer_mod, PeerModule},
{priv_data_dir, PrivDataDir},
{load_schema, true},
{start_autocluster, true},
@@ -1744,14 +1736,14 @@ t_cluster_group(Config) ->
begin
Nodes =
[_N1, N2 | _] = [
- emqx_common_test_helpers:start_slave(Name, Opts)
+ emqx_common_test_helpers:start_peer(Name, Opts)
|| {Name, Opts} <- Cluster
],
on_exit(fun() ->
emqx_utils:pmap(
fun(N) ->
ct:pal("stopping ~p", [N]),
- ok = emqx_common_test_helpers:stop_slave(N)
+ ok = emqx_common_test_helpers:stop_peer(N)
end,
Nodes
)
@@ -1827,10 +1819,10 @@ t_node_joins_existing_cluster(Config) ->
begin
[{Name1, Opts1}, {Name2, Opts2} | _] = Cluster,
ct:pal("starting ~p", [Name1]),
- N1 = emqx_common_test_helpers:start_slave(Name1, Opts1),
+ N1 = emqx_common_test_helpers:start_peer(Name1, Opts1),
on_exit(fun() ->
ct:pal("stopping ~p", [N1]),
- ok = emqx_common_test_helpers:stop_slave(N1)
+ ok = emqx_common_test_helpers:stop_peer(N1)
end),
{{ok, _}, {ok, _}} =
?wait_async_action(
@@ -1870,10 +1862,10 @@ t_node_joins_existing_cluster(Config) ->
30_000
),
ct:pal("starting ~p", [Name2]),
- N2 = emqx_common_test_helpers:start_slave(Name2, Opts2),
+ N2 = emqx_common_test_helpers:start_peer(Name2, Opts2),
on_exit(fun() ->
ct:pal("stopping ~p", [N2]),
- ok = emqx_common_test_helpers:stop_slave(N2)
+ ok = emqx_common_test_helpers:stop_peer(N2)
end),
Nodes = [N1, N2],
wait_for_cluster_rpc(N2),
@@ -1963,7 +1955,7 @@ t_cluster_node_down(Config) ->
lists:map(
fun({Name, Opts}) ->
ct:pal("starting ~p", [Name]),
- emqx_common_test_helpers:start_slave(Name, Opts)
+ emqx_common_test_helpers:start_peer(Name, Opts)
end,
Cluster
),
@@ -1971,7 +1963,7 @@ t_cluster_node_down(Config) ->
emqx_utils:pmap(
fun(N) ->
ct:pal("stopping ~p", [N]),
- ok = emqx_common_test_helpers:stop_slave(N)
+ ok = emqx_common_test_helpers:stop_peer(N)
end,
Nodes
)
@@ -2016,7 +2008,7 @@ t_cluster_node_down(Config) ->
{TId, Pid} = start_async_publisher(Config, KafkaTopic),
ct:pal("stopping node ~p", [N1]),
- ok = emqx_common_test_helpers:stop_slave(N1),
+ ok = emqx_common_test_helpers:stop_peer(N1),
%% Give some time for the consumers in remaining node to
%% rebalance.
diff --git a/apps/emqx_bridge_pulsar/test/emqx_bridge_pulsar_impl_producer_SUITE.erl b/apps/emqx_bridge_pulsar/test/emqx_bridge_pulsar_impl_producer_SUITE.erl
index 53c883297..6b5bc2896 100644
--- a/apps/emqx_bridge_pulsar/test/emqx_bridge_pulsar_impl_producer_SUITE.erl
+++ b/apps/emqx_bridge_pulsar/test/emqx_bridge_pulsar_impl_producer_SUITE.erl
@@ -517,19 +517,11 @@ try_decode_json(Payload) ->
cluster(Config) ->
PrivDataDir = ?config(priv_dir, Config),
- PeerModule =
- case os:getenv("IS_CI") of
- false ->
- slave;
- _ ->
- ct_slave
- end,
Cluster = emqx_common_test_helpers:emqx_cluster(
[core, core],
[
{apps, [emqx_conf] ++ ?APPS ++ [pulsar]},
{listener_ports, []},
- {peer_mod, PeerModule},
{priv_data_dir, PrivDataDir},
{load_schema, true},
{start_autocluster, true},
@@ -551,7 +543,7 @@ cluster(Config) ->
start_cluster(Cluster) ->
Nodes =
[
- emqx_common_test_helpers:start_slave(Name, Opts)
+ emqx_common_test_helpers:start_peer(Name, Opts)
|| {Name, Opts} <- Cluster
],
NumNodes = length(Nodes),
@@ -559,7 +551,7 @@ start_cluster(Cluster) ->
emqx_utils:pmap(
fun(N) ->
ct:pal("stopping ~p", [N]),
- ok = emqx_common_test_helpers:stop_slave(N)
+ ok = emqx_common_test_helpers:stop_peer(N)
end,
Nodes
)
diff --git a/apps/emqx_conf/test/emqx_conf_app_SUITE.erl b/apps/emqx_conf/test/emqx_conf_app_SUITE.erl
index 2e5da3d44..155489af7 100644
--- a/apps/emqx_conf/test/emqx_conf_app_SUITE.erl
+++ b/apps/emqx_conf/test/emqx_conf_app_SUITE.erl
@@ -222,16 +222,16 @@ assert_config_load_done(Nodes) ->
).
stop_cluster(Nodes) ->
- emqx_utils:pmap(fun emqx_common_test_helpers:stop_slave/1, Nodes).
+ emqx_utils:pmap(fun emqx_common_test_helpers:stop_peer/1, Nodes).
start_cluster(Specs) ->
- [emqx_common_test_helpers:start_slave(Name, Opts) || {Name, Opts} <- Specs].
+ [emqx_common_test_helpers:start_peer(Name, Opts) || {Name, Opts} <- Specs].
start_cluster_async(Specs) ->
[
begin
Opts1 = maps:remove(join_to, Opts),
- spawn_link(fun() -> emqx_common_test_helpers:start_slave(Name, Opts1) end),
+ spawn_link(fun() -> emqx_common_test_helpers:start_peer(Name, Opts1) end),
timer:sleep(7_000)
end
|| {Name, Opts} <- Specs
diff --git a/apps/emqx_eviction_agent/test/emqx_eviction_agent_SUITE.erl b/apps/emqx_eviction_agent/test/emqx_eviction_agent_SUITE.erl
index bf2865a78..2dd50ce2b 100644
--- a/apps/emqx_eviction_agent/test/emqx_eviction_agent_SUITE.erl
+++ b/apps/emqx_eviction_agent/test/emqx_eviction_agent_SUITE.erl
@@ -50,9 +50,9 @@ end_per_suite(Config) ->
init_per_testcase(Case, Config) ->
_ = emqx_eviction_agent:disable(test_eviction),
ok = snabbkaffe:start_trace(),
- start_slave(Case, Config).
+ start_peer(Case, Config).
-start_slave(t_explicit_session_takeover, Config) ->
+start_peer(t_explicit_session_takeover, Config) ->
NodeNames =
[
t_explicit_session_takeover_donor,
@@ -65,19 +65,19 @@ start_slave(t_explicit_session_takeover, Config) ->
),
ok = snabbkaffe:start_trace(),
[{evacuate_nodes, ClusterNodes} | Config];
-start_slave(_Case, Config) ->
+start_peer(_Case, Config) ->
Config.
end_per_testcase(TestCase, Config) ->
emqx_eviction_agent:disable(test_eviction),
ok = snabbkaffe:stop(),
- stop_slave(TestCase, Config).
+ stop_peer(TestCase, Config).
-stop_slave(t_explicit_session_takeover, Config) ->
+stop_peer(t_explicit_session_takeover, Config) ->
emqx_eviction_agent_test_helpers:stop_cluster(
?config(evacuate_nodes, Config)
);
-stop_slave(_Case, _Config) ->
+stop_peer(_Case, _Config) ->
ok.
%%--------------------------------------------------------------------
diff --git a/apps/emqx_license/test/emqx_license_SUITE.erl b/apps/emqx_license/test/emqx_license_SUITE.erl
index 4818ad9e6..fcbced33a 100644
--- a/apps/emqx_license/test/emqx_license_SUITE.erl
+++ b/apps/emqx_license/test/emqx_license_SUITE.erl
@@ -112,7 +112,7 @@ setup_test(TestCase, Config) when
end}
]
),
- Nodes = [emqx_common_test_helpers:start_slave(Name, Opts) || {Name, Opts} <- Cluster],
+ Nodes = [emqx_common_test_helpers:start_peer(Name, Opts) || {Name, Opts} <- Cluster],
[{nodes, Nodes}, {cluster, Cluster}, {old_license, LicenseKey}];
setup_test(_TestCase, _Config) ->
[].
diff --git a/apps/emqx_management/test/emqx_mgmt_api_SUITE.erl b/apps/emqx_management/test/emqx_mgmt_api_SUITE.erl
index 0e63b38ab..ade8d3171 100644
--- a/apps/emqx_management/test/emqx_mgmt_api_SUITE.erl
+++ b/apps/emqx_management/test/emqx_mgmt_api_SUITE.erl
@@ -42,8 +42,8 @@ t_cluster_query(_Config) ->
ct:timetrap({seconds, 120}),
snabbkaffe:fix_ct_logging(),
[{Name, Opts}, {Name1, Opts1}] = cluster_specs(),
- Node1 = emqx_common_test_helpers:start_slave(Name, Opts),
- Node2 = emqx_common_test_helpers:start_slave(Name1, Opts1),
+ Node1 = emqx_common_test_helpers:start_peer(Name, Opts),
+ Node2 = emqx_common_test_helpers:start_peer(Name1, Opts1),
try
process_flag(trap_exit, true),
ClientLs1 = [start_emqtt_client(Node1, I, 2883) || I <- lists:seq(1, 10)],
@@ -168,8 +168,8 @@ t_cluster_query(_Config) ->
_ = lists:foreach(fun(C) -> emqtt:disconnect(C) end, ClientLs1),
_ = lists:foreach(fun(C) -> emqtt:disconnect(C) end, ClientLs2)
after
- emqx_common_test_helpers:stop_slave(Node1),
- emqx_common_test_helpers:stop_slave(Node2)
+ emqx_common_test_helpers:stop_peer(Node1),
+ emqx_common_test_helpers:stop_peer(Node2)
end,
ok.
diff --git a/apps/emqx_management/test/emqx_mgmt_api_cluster_SUITE.erl b/apps/emqx_management/test/emqx_mgmt_api_cluster_SUITE.erl
index abe201217..6e33c4001 100644
--- a/apps/emqx_management/test/emqx_mgmt_api_cluster_SUITE.erl
+++ b/apps/emqx_management/test/emqx_mgmt_api_cluster_SUITE.erl
@@ -54,8 +54,6 @@ t_cluster_topology_api_empty_resp(_) ->
).
t_cluster_topology_api_replicants(Config) ->
- %% some time to stabilize
- timer:sleep(3000),
[Core1, Core2, Replicant] = _NodesList = ?config(cluster, Config),
{200, Core1Resp} = rpc:call(Core1, emqx_mgmt_api_cluster, cluster_topology, [get, #{}]),
{200, Core2Resp} = rpc:call(Core2, emqx_mgmt_api_cluster, cluster_topology, [get, #{}]),
diff --git a/apps/emqx_management/test/emqx_mgmt_api_listeners_SUITE.erl b/apps/emqx_management/test/emqx_mgmt_api_listeners_SUITE.erl
index 53579406f..cbcbfdd3f 100644
--- a/apps/emqx_management/test/emqx_mgmt_api_listeners_SUITE.erl
+++ b/apps/emqx_management/test/emqx_mgmt_api_listeners_SUITE.erl
@@ -194,8 +194,8 @@ t_api_listeners_list_not_ready(Config) when is_list(Config) ->
snabbkaffe:fix_ct_logging(),
Cluster = [{Name, Opts}, {Name1, Opts1}] = cluster([core, core]),
ct:pal("Starting ~p", [Cluster]),
- Node1 = emqx_common_test_helpers:start_slave(Name, Opts),
- Node2 = emqx_common_test_helpers:start_slave(Name1, Opts1),
+ Node1 = emqx_common_test_helpers:start_peer(Name, Opts),
+ Node2 = emqx_common_test_helpers:start_peer(Name1, Opts1),
try
L1 = get_tcp_listeners(Node1),
@@ -214,8 +214,8 @@ t_api_listeners_list_not_ready(Config) when is_list(Config) ->
?assert(length(L1) > length(L2), Comment),
?assertEqual(length(L2), length(L3), Comment)
after
- emqx_common_test_helpers:stop_slave(Node1),
- emqx_common_test_helpers:stop_slave(Node2)
+ emqx_common_test_helpers:stop_peer(Node1),
+ emqx_common_test_helpers:stop_peer(Node2)
end.
t_clear_certs(Config) when is_list(Config) ->
diff --git a/apps/emqx_management/test/emqx_mgmt_api_nodes_SUITE.erl b/apps/emqx_management/test/emqx_mgmt_api_nodes_SUITE.erl
index 8d98b974d..63754d437 100644
--- a/apps/emqx_management/test/emqx_mgmt_api_nodes_SUITE.erl
+++ b/apps/emqx_management/test/emqx_mgmt_api_nodes_SUITE.erl
@@ -129,8 +129,8 @@ t_multiple_nodes_api(_) ->
Seq2 = list_to_atom(atom_to_list(?MODULE) ++ "2"),
Cluster = [{Name, Opts}, {Name1, Opts1}] = cluster([{core, Seq1}, {core, Seq2}]),
ct:pal("Starting ~p", [Cluster]),
- Node1 = emqx_common_test_helpers:start_slave(Name, Opts),
- Node2 = emqx_common_test_helpers:start_slave(Name1, Opts1),
+ Node1 = emqx_common_test_helpers:start_peer(Name, Opts),
+ Node2 = emqx_common_test_helpers:start_peer(Name1, Opts1),
try
{200, NodesList} = rpc:call(Node1, emqx_mgmt_api_nodes, nodes, [get, #{}]),
All = [Node1, Node2],
@@ -148,8 +148,8 @@ t_multiple_nodes_api(_) ->
]),
?assertMatch(#{node := Node1}, Node11)
after
- emqx_common_test_helpers:stop_slave(Node1),
- emqx_common_test_helpers:stop_slave(Node2)
+ emqx_common_test_helpers:stop_peer(Node1),
+ emqx_common_test_helpers:stop_peer(Node2)
end,
ok.
diff --git a/apps/emqx_management/test/emqx_mgmt_api_topics_SUITE.erl b/apps/emqx_management/test/emqx_mgmt_api_topics_SUITE.erl
index 854f1133b..40ca7aa91 100644
--- a/apps/emqx_management/test/emqx_mgmt_api_topics_SUITE.erl
+++ b/apps/emqx_management/test/emqx_mgmt_api_topics_SUITE.erl
@@ -27,12 +27,12 @@ all() ->
init_per_suite(Config) ->
emqx_mgmt_api_test_util:init_suite(),
- Slave = emqx_common_test_helpers:start_slave(some_node, []),
- [{slave, Slave} | Config].
+ Peer = emqx_common_test_helpers:start_peer(node1, []),
+ [{peer, Peer} | Config].
end_per_suite(Config) ->
- Slave = ?config(slave, Config),
- emqx_common_test_helpers:stop_slave(Slave),
+ Peer = ?config(peer, Config),
+ emqx_common_test_helpers:stop_peer(Peer),
mria:clear_table(?ROUTE_TAB),
emqx_mgmt_api_test_util:end_suite().
@@ -80,18 +80,18 @@ t_nodes_api(Config) ->
%% get topics/:topic
%% We add another route here to ensure that the response handles
%% multiple routes for a single topic
- Slave = ?config(slave, Config),
- ok = emqx_router:add_route(Topic, Slave),
+ Peer = ?config(peer, Config),
+ ok = emqx_router:add_route(Topic, Peer),
RoutePath = emqx_mgmt_api_test_util:api_path(["topics", Topic]),
{ok, RouteResponse} = emqx_mgmt_api_test_util:request_api(get, RoutePath),
- ok = emqx_router:delete_route(Topic, Slave),
+ ok = emqx_router:delete_route(Topic, Peer),
[
#{<<"topic">> := Topic, <<"node">> := Node1},
#{<<"topic">> := Topic, <<"node">> := Node2}
] = emqx_utils_json:decode(RouteResponse, [return_maps]),
- ?assertEqual(lists:usort([Node, atom_to_binary(Slave)]), lists:usort([Node1, Node2])),
+ ?assertEqual(lists:usort([Node, atom_to_binary(Peer)]), lists:usort([Node1, Node2])),
ok = emqtt:stop(Client).
diff --git a/apps/emqx_node_rebalance/test/emqx_node_rebalance_SUITE.erl b/apps/emqx_node_rebalance/test/emqx_node_rebalance_SUITE.erl
index d996719fb..54e810c9d 100644
--- a/apps/emqx_node_rebalance/test/emqx_node_rebalance_SUITE.erl
+++ b/apps/emqx_node_rebalance/test/emqx_node_rebalance_SUITE.erl
@@ -136,7 +136,7 @@ t_rebalance_node_crash(Config) ->
?assertWaitEvent(
begin
ok = rpc:call(DonorNode, emqx_node_rebalance, start, [Opts]),
- emqx_common_test_helpers:stop_slave(RecipientNode)
+ emqx_common_test_helpers:stop_peer(RecipientNode)
end,
#{?snk_kind := emqx_node_rebalance_started},
1000
diff --git a/apps/emqx_plugins/test/emqx_plugins_SUITE.erl b/apps/emqx_plugins/test/emqx_plugins_SUITE.erl
index b0a47a6a0..b83597279 100644
--- a/apps/emqx_plugins/test/emqx_plugins_SUITE.erl
+++ b/apps/emqx_plugins/test/emqx_plugins_SUITE.erl
@@ -628,11 +628,11 @@ group_t_copy_plugin_to_a_new_node({init, Config}) ->
load_schema => false
}
),
- CopyFromNode = emqx_common_test_helpers:start_slave(
+ CopyFromNode = emqx_common_test_helpers:start_peer(
CopyFrom, maps:remove(join_to, CopyFromOpts)
),
ok = rpc:call(CopyFromNode, emqx_plugins, put_config, [install_dir, FromInstallDir]),
- CopyToNode = emqx_common_test_helpers:start_slave(CopyTo, maps:remove(join_to, CopyToOpts)),
+ CopyToNode = emqx_common_test_helpers:start_peer(CopyTo, maps:remove(join_to, CopyToOpts)),
ok = rpc:call(CopyToNode, emqx_plugins, put_config, [install_dir, ToInstallDir]),
NameVsn = filename:basename(Package, ?PACKAGE_SUFFIX),
ok = rpc:call(CopyFromNode, emqx_plugins, ensure_installed, [NameVsn]),
@@ -662,8 +662,8 @@ group_t_copy_plugin_to_a_new_node({'end', Config}) ->
ok = rpc:call(CopyToNode, emqx_config, delete_override_conf_files, []),
rpc:call(CopyToNode, ekka, leave, []),
rpc:call(CopyFromNode, ekka, leave, []),
- ok = emqx_common_test_helpers:stop_slave(CopyToNode),
- ok = emqx_common_test_helpers:stop_slave(CopyFromNode),
+ ok = emqx_common_test_helpers:stop_peer(CopyToNode),
+ ok = emqx_common_test_helpers:stop_peer(CopyFromNode),
ok = file:del_dir_r(proplists:get_value(to_install_dir, Config)),
ok = file:del_dir_r(proplists:get_value(from_install_dir, Config));
group_t_copy_plugin_to_a_new_node(Config) ->
@@ -737,7 +737,6 @@ group_t_copy_plugin_to_a_new_node_single_node({init, Config}) ->
end,
priv_data_dir => PrivDataDir,
schema_mod => emqx_conf_schema,
- peer_mod => slave,
load_schema => true
}
),
@@ -751,7 +750,7 @@ group_t_copy_plugin_to_a_new_node_single_node({init, Config}) ->
];
group_t_copy_plugin_to_a_new_node_single_node({'end', Config}) ->
CopyToNode = proplists:get_value(copy_to_node_name, Config),
- ok = emqx_common_test_helpers:stop_slave(CopyToNode),
+ ok = emqx_common_test_helpers:stop_peer(CopyToNode),
ok = file:del_dir_r(proplists:get_value(to_install_dir, Config)),
ok;
group_t_copy_plugin_to_a_new_node_single_node(Config) ->
@@ -762,7 +761,7 @@ group_t_copy_plugin_to_a_new_node_single_node(Config) ->
%% Start the node for the first time. The plugin should start
%% successfully even if it's not extracted yet. Simply starting
%% the node would crash if not working properly.
- CopyToNode = emqx_common_test_helpers:start_slave(CopyTo, CopyToOpts),
+ CopyToNode = emqx_common_test_helpers:start_peer(CopyTo, CopyToOpts),
ct:pal("~p config:\n ~p", [
CopyToNode, erpc:call(CopyToNode, emqx_plugins, get_config, [[], #{}])
]),
@@ -805,11 +804,10 @@ group_t_cluster_leave({init, Config}) ->
end,
priv_data_dir => PrivDataDir,
schema_mod => emqx_conf_schema,
- peer_mod => slave,
load_schema => true
}
),
- Nodes = [emqx_common_test_helpers:start_slave(Name, Opts) || {Name, Opts} <- Cluster],
+ Nodes = [emqx_common_test_helpers:start_peer(Name, Opts) || {Name, Opts} <- Cluster],
[
{to_install_dir, ToInstallDir},
{cluster, Cluster},
@@ -820,7 +818,7 @@ group_t_cluster_leave({init, Config}) ->
];
group_t_cluster_leave({'end', Config}) ->
Nodes = proplists:get_value(nodes, Config),
- [ok = emqx_common_test_helpers:stop_slave(N) || N <- Nodes],
+ [ok = emqx_common_test_helpers:stop_peer(N) || N <- Nodes],
ok = file:del_dir_r(proplists:get_value(to_install_dir, Config)),
ok;
group_t_cluster_leave(Config) ->
diff --git a/apps/emqx_schema_registry/test/emqx_schema_registry_SUITE.erl b/apps/emqx_schema_registry/test/emqx_schema_registry_SUITE.erl
index 7aea09457..f316f2991 100644
--- a/apps/emqx_schema_registry/test/emqx_schema_registry_SUITE.erl
+++ b/apps/emqx_schema_registry/test/emqx_schema_registry_SUITE.erl
@@ -348,19 +348,11 @@ receive_published(Line) ->
cluster(Config) ->
PrivDataDir = ?config(priv_dir, Config),
- PeerModule =
- case os:getenv("IS_CI") of
- false ->
- slave;
- _ ->
- ct_slave
- end,
Cluster = emqx_common_test_helpers:emqx_cluster(
[core, core],
[
{apps, ?APPS},
{listener_ports, []},
- {peer_mod, PeerModule},
{priv_data_dir, PrivDataDir},
{load_schema, true},
{start_autocluster, true},
@@ -382,7 +374,7 @@ cluster(Config) ->
start_cluster(Cluster) ->
Nodes = [
- emqx_common_test_helpers:start_slave(Name, Opts)
+ emqx_common_test_helpers:start_peer(Name, Opts)
|| {Name, Opts} <- Cluster
],
NumNodes = length(Nodes),
@@ -390,7 +382,7 @@ start_cluster(Cluster) ->
emqx_utils:pmap(
fun(N) ->
ct:pal("stopping ~p", [N]),
- ok = emqx_common_test_helpers:stop_slave(N)
+ ok = emqx_common_test_helpers:stop_peer(N)
end,
Nodes
)
diff --git a/apps/emqx_telemetry/test/emqx_telemetry_SUITE.erl b/apps/emqx_telemetry/test/emqx_telemetry_SUITE.erl
index 8f47e8290..396a7fb37 100644
--- a/apps/emqx_telemetry/test/emqx_telemetry_SUITE.erl
+++ b/apps/emqx_telemetry/test/emqx_telemetry_SUITE.erl
@@ -154,7 +154,7 @@ init_per_testcase(t_exhook_info, Config) ->
emqx_common_test_helpers:start_apps([emqx_exhook]),
Config;
init_per_testcase(t_cluster_uuid, Config) ->
- Node = start_slave(n1),
+ Node = start_peer(n1),
[{n1, Node} | Config];
init_per_testcase(t_uuid_restored_from_file, Config) ->
Config;
@@ -210,7 +210,7 @@ end_per_testcase(t_exhook_info, _Config) ->
ok;
end_per_testcase(t_cluster_uuid, Config) ->
Node = proplists:get_value(n1, Config),
- ok = stop_slave(Node);
+ ok = stop_peer(Node);
end_per_testcase(t_num_clients, Config) ->
ok = snabbkaffe:stop(),
Config;
@@ -782,7 +782,7 @@ find_gen_rpc_port() ->
{ok, {_, Port}} = inet:sockname(EPort),
Port.
-start_slave(Name) ->
+start_peer(Name) ->
Port = find_gen_rpc_port(),
TestNode = node(),
Handler =
@@ -811,11 +811,9 @@ start_slave(Name) ->
apps => [emqx, emqx_conf, emqx_retainer, emqx_modules, emqx_telemetry]
},
- emqx_common_test_helpers:start_slave(Name, Opts).
+ emqx_common_test_helpers:start_peer(Name, Opts).
-stop_slave(Node) ->
- % This line don't work!!
- %emqx_cluster_rpc:fast_forward_to_commit(Node, 100),
+stop_peer(Node) ->
rpc:call(Node, ?MODULE, leave_cluster, []),
ok = emqx_cth_peer:stop(Node),
?assertEqual([node()], mria:running_nodes()),
diff --git a/rel/i18n/emqx_schema.hocon b/rel/i18n/emqx_schema.hocon
index 711b5c2bc..ea2847341 100644
--- a/rel/i18n/emqx_schema.hocon
+++ b/rel/i18n/emqx_schema.hocon
@@ -1386,7 +1386,7 @@ However it's no longer useful because the shared-subscrption messages in a expir
base_listener_enable_authn.desc:
"""Set true
(default) to enable client authentication on this listener, the authentication
process goes through the configured authentication chain.
-When set to false
to allow any clients with or without authentication information such as username or password to log in.
+When set to false
, any client (with or without username/password) is allowed to connect.
When set to quick_deny_anonymous
, it behaves like when set to true
, but clients will be
denied immediately without going through any authenticators if username
is not provided. This is useful to fence off
anonymous clients early."""