test: fix a flacky test case

src/emqx.appup.src

@@ -3,7 +3,8 @@
   [{"4.3.2",
     [{load_module,emqx_http_lib,brutal_purge,soft_purge,[]},
      {load_module,emqx_channel,brutal_purge,soft_purge,[]},
-     {load_module,emqx_app,brutal_purge,soft_purge,[]}]},
+     {load_module,emqx_app,brutal_purge,soft_purge,[]},
+     {load_module,emqx_connection,brutal_purge,soft_purge,[]}]},
    {"4.3.1",
     [{load_module,emqx_ws_connection,brutal_purge,soft_purge,[]},
      {load_module,emqx_connection,brutal_purge,soft_purge,[]},
@@ -36,7 +37,8 @@
   [{"4.3.2",
     [{load_module,emqx_http_lib,brutal_purge,soft_purge,[]},
      {load_module,emqx_channel,brutal_purge,soft_purge,[]},
-     {load_module,emqx_app,brutal_purge,soft_purge,[]}]},
+     {load_module,emqx_app,brutal_purge,soft_purge,[]},
+     {load_module,emqx_connection,brutal_purge,soft_purge,[]}]},
    {"4.3.1",
     [{load_module,emqx_ws_connection,brutal_purge,soft_purge,[]},
      {load_module,emqx_connection,brutal_purge,soft_purge,[]},

src/emqx_connection.erl

@@ -349,6 +349,7 @@ ensure_stats_timer(_Timeout, State) -> State.
 
 -compile({inline, [cancel_stats_timer/1]}).
 cancel_stats_timer(State = #state{stats_timer = TRef}) when is_reference(TRef) ->
+    ?tp(debug, cancel_stats_timer, #{}),
     ok = emqx_misc:cancel_timer(TRef),
     State#state{stats_timer = undefined};
 cancel_stats_timer(State) -> State.

test/emqx_mqtt_protocol_v5_SUITE.erl

@@ -22,6 +22,7 @@
 -include_lib("emqx/include/emqx.hrl").
 -include_lib("emqx/include/emqx_mqtt.hrl").
 -include_lib("eunit/include/eunit.hrl").
+-include_lib("snabbkaffe/include/snabbkaffe.hrl").
 
 -import(lists, [nth/2]).
 
@@ -45,6 +46,19 @@ end_per_suite(_Config) ->
     ok = meck:unload(emqtt),
     emqx_ct_helpers:stop_apps([]).
 
+init_per_testcase(TestCase, Config) ->
+    case erlang:function_exported(?MODULE, TestCase, 2) of
+        true -> ?MODULE:TestCase(init, Config);
+        _ -> Config
+    end.
+
+end_per_testcase(TestCase, Config) ->
+    case erlang:function_exported(?MODULE, TestCase, 2) of
+        true -> ?MODULE:TestCase('end', Config);
+        false -> ok
+    end,
+    Config.
+
 %%--------------------------------------------------------------------
 %% Helpers
 %%--------------------------------------------------------------------
@@ -274,16 +288,25 @@ t_connect_limit_timeout(_) ->
     emqx_zone:set_env(external, publish_limit, undefined),
     meck:unload(proplists).
 
-t_connect_emit_stats_timeout(_) ->
+t_connect_emit_stats_timeout(init, Config) ->
-    IdleTimeout = 2000,
+    NewIdleTimeout = 1000,
-    emqx_zone:set_env(external, idle_timeout, IdleTimeout),
+    OldIdleTimeout = emqx_zone:get_env(external, idle_timeout),
+    emqx_zone:set_env(external, idle_timeout, NewIdleTimeout),
+    ok = snabbkaffe:start_trace(),
+    [{idle_timeout, NewIdleTimeout}, {old_idle_timeout, OldIdleTimeout} | Config];
+t_connect_emit_stats_timeout('end', Config) ->
+    snabbkaffe:stop(),
+    {_, OldIdleTimeout} = lists:keyfind(old_idle_timeout, 1, Config),
+    emqx_zone:set_env(external, idle_timeout, OldIdleTimeout),
+    ok.
 
+t_connect_emit_stats_timeout(Config) ->
+    {_, IdleTimeout} = lists:keyfind(idle_timeout, 1, Config),
     {ok, Client} = emqtt:start_link([{proto_ver, v5},{keepalive, 60}]),
     {ok, _} = emqtt:connect(Client),
     [ClientPid] = emqx_cm:lookup_channels(client_info(clientid, Client)),
-
     ?assert(is_reference(emqx_connection:info(stats_timer, sys:get_state(ClientPid)))),
-    timer:sleep(IdleTimeout),
+    ?block_until(#{?snk_kind := cancel_stats_timer}, IdleTimeout * 2, _BackInTime = 0),
     ?assertEqual(undefined, emqx_connection:info(stats_timer, sys:get_state(ClientPid))),
     ok = emqtt:disconnect(Client).