Merge pull request #12074 from zmstone/1201-user-peer-module-for-all

test: replace 'slave' and 'ct_slave' with 'peer'

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] ->

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) ->

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).
 

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
         )

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.

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
         )

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

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.
 
 %%--------------------------------------------------------------------

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) ->
     [].

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.
 

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, #{}]),

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) ->

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.
 

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).
 

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

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) ->

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
         )

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()),

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 <code>true</code> (default) to enable client authentication on this listener, the authentication
 process goes through the configured authentication chain.
-When set to <code>false</code> to allow any clients with or without authentication information such as username or password to log in.
+When set to <code>false</code>, any client (with or without username/password) is allowed to connect.
 When set to <code>quick_deny_anonymous</code>, it behaves like when set to <code>true</code>, but clients will be
 denied immediately without going through any authenticators if <code>username</code> is not provided. This is useful to fence off
 anonymous clients early."""