diff --git a/apps/emqx/integration_test/emqx_persistent_session_ds_SUITE.erl b/apps/emqx/integration_test/emqx_persistent_session_ds_SUITE.erl
index 265ec02b9..6f064940e 100644
--- a/apps/emqx/integration_test/emqx_persistent_session_ds_SUITE.erl
+++ b/apps/emqx/integration_test/emqx_persistent_session_ds_SUITE.erl
@@ -40,7 +40,7 @@ init_per_testcase(TestCase, Config) when
     Cluster = cluster(#{n => 1}),
     ClusterOpts = #{work_dir => emqx_cth_suite:work_dir(TestCase, Config)},
     NodeSpecs = emqx_cth_cluster:mk_nodespecs(Cluster, ClusterOpts),
-    Nodes = emqx_cth_cluster:start(Cluster, ClusterOpts),
+    Nodes = emqx_cth_cluster:start(NodeSpecs),
     [
         {cluster, Cluster},
         {node_specs, NodeSpecs},
@@ -116,24 +116,8 @@ start_client(Opts0 = #{}) ->
 
 restart_node(Node, NodeSpec) ->
     ?tp(will_restart_node, #{}),
-    ?tp(notice, "restarting node", #{node => Node}),
-    true = monitor_node(Node, true),
-    ok = erpc:call(Node, init, restart, []),
-    receive
-        {nodedown, Node} ->
-            ok
-    after 10_000 ->
-        ct:fail("node ~p didn't stop", [Node])
-    end,
-    ?tp(notice, "waiting for nodeup", #{node => Node}),
+    emqx_cth_cluster:restart(Node, NodeSpec),
     wait_nodeup(Node),
-    wait_gen_rpc_down(NodeSpec),
-    ?tp(notice, "restarting apps", #{node => Node}),
-    Apps = maps:get(apps, NodeSpec),
-    ok = erpc:call(Node, emqx_cth_suite, load_apps, [Apps]),
-    _ = erpc:call(Node, emqx_cth_suite, start_apps, [Apps, NodeSpec]),
-    ok = snabbkaffe:forward_trace(Node),
-    ?tp(notice, "node restarted", #{node => Node}),
     ?tp(restarted_node, #{}),
     ok.
 
diff --git a/apps/emqx/test/emqx_cth_cluster.erl b/apps/emqx/test/emqx_cth_cluster.erl
index 49212ba97..029907f57 100644
--- a/apps/emqx/test/emqx_cth_cluster.erl
+++ b/apps/emqx/test/emqx_cth_cluster.erl
@@ -38,7 +38,7 @@
 %%    in `end_per_suite/1` or `end_per_group/2`) with the result from step 2.
 -module(emqx_cth_cluster).
 
--export([start/2]).
+-export([start/1, start/2, restart/2]).
 -export([stop/1, stop_node/1]).
 
 -export([start_bare_nodes/1, start_bare_nodes/2]).
@@ -109,7 +109,10 @@ when
     }.
 start(Nodes, ClusterOpts) ->
     NodeSpecs = mk_nodespecs(Nodes, ClusterOpts),
-    ct:pal("Starting cluster:\n  ~p", [NodeSpecs]),
+    start(NodeSpecs).
+
+start(NodeSpecs) ->
+    ct:pal("(Re)starting nodes:\n  ~p", [NodeSpecs]),
     % 1. Start bare nodes with only basic applications running
     ok = start_nodes_init(NodeSpecs, ?TIMEOUT_NODE_START_MS),
     % 2. Start applications needed to enable clustering
@@ -121,6 +124,11 @@ start(Nodes, ClusterOpts) ->
     _ = emqx_utils:pmap(fun run_node_phase_apps/1, NodeSpecs, ?TIMEOUT_APPS_START_MS),
     [Node || #{name := Node} <- NodeSpecs].
 
+restart(Node, Spec) ->
+    ct:pal("Stopping peer node ~p", [Node]),
+    ok = emqx_cth_peer:stop(Node),
+    start([Spec#{boot_type => restart}]).
+
 mk_nodespecs(Nodes, ClusterOpts) ->
     NodeSpecs = lists:zipwith(
         fun(N, {Name, Opts}) -> mk_init_nodespec(N, Name, Opts, ClusterOpts) end,
@@ -358,8 +366,12 @@ start_apps(Node, #{apps := Apps} = Spec) ->
     ok.
 
 suite_opts(Spec) ->
-    maps:with([work_dir], Spec).
+    maps:with([work_dir, boot_type], Spec).
 
+maybe_join_cluster(_Node, #{boot_type := restart}) ->
+    %% when restart, the node should already be in the cluster
+    %% hence no need to (re)join
+    ok;
 maybe_join_cluster(_Node, #{role := replicant}) ->
     ok;
 maybe_join_cluster(Node, Spec) ->
diff --git a/apps/emqx/test/emqx_cth_suite.erl b/apps/emqx/test/emqx_cth_suite.erl
index 401d4f59d..5e91b92c9 100644
--- a/apps/emqx/test/emqx_cth_suite.erl
+++ b/apps/emqx/test/emqx_cth_suite.erl
@@ -453,6 +453,9 @@ stop_apps(Apps) ->
 
 %%
 
+verify_clean_suite_state(#{boot_type := restart}) ->
+    %% when testing node restart, we do not need to verify clean state
+    ok;
 verify_clean_suite_state(#{work_dir := WorkDir}) ->
     {ok, []} = file:list_dir(WorkDir),
     false = emqx_schema_hooks:any_injections(),