test: trying to make tests more stable

apps/emqx/test/emqx_common_test_helpers.erl

@@ -75,7 +75,8 @@
 -export([with_mock/4]).
 -export([
     on_exit/1,
-    call_janitor/0
+    call_janitor/0,
+    call_janitor/1
 ]).
 
 %% Toxiproxy API
@@ -1078,8 +1079,11 @@ latency_up_proxy(off, Name, ProxyHost, ProxyPort) ->
 %% stop the janitor gracefully to ensure proper cleanup order and less
 %% noise in the logs.
 call_janitor() ->
+    call_janitor(15_000).
+
+call_janitor(Timeout) ->
     Janitor = get_or_spawn_janitor(),
-    ok = emqx_test_janitor:stop(Janitor),
+    ok = emqx_test_janitor:stop(Janitor, Timeout),
     ok.
 
 get_or_spawn_janitor() ->

apps/emqx/test/emqx_test_janitor.erl

@@ -31,6 +31,7 @@
 -export([
     start_link/0,
     stop/1,
+    stop/2,
     push_on_exit_callback/2
 ]).
 
@@ -42,7 +43,10 @@ start_link() ->
     gen_server:start_link(?MODULE, self(), []).
 
 stop(Server) ->
-    gen_server:call(Server, terminate).
+    stop(Server, 15_000).
+
+stop(Server, Timeout) ->
+    gen_server:call(Server, terminate, Timeout).
 
 push_on_exit_callback(Server, Callback) when is_function(Callback, 0) ->
     gen_server:call(Server, {push, Callback}).
@@ -56,7 +60,7 @@ init(Parent) ->
     {ok, #{callbacks => [], owner => Parent}}.
 
 terminate(_Reason, #{callbacks := Callbacks}) ->
-    lists:foreach(fun(Fun) -> Fun() end, Callbacks).
+    lists:foreach(fun(Fun) -> catch Fun() end, Callbacks).
 
 handle_call({push, Callback}, _From, State = #{callbacks := Callbacks}) ->
     {reply, ok, State#{callbacks := [Callback | Callbacks]}};

lib-ee/emqx_ee_bridge/test/emqx_bridge_impl_kafka_consumer_SUITE.erl

@@ -360,7 +360,6 @@ common_init_per_testcase(TestCase, Config0) ->
 end_per_testcase(_Testcase, Config) ->
     case proplists:get_bool(skip_does_not_apply, Config) of
         true ->
-            ok = snabbkaffe:stop(),
             ok;
         false ->
             ProxyHost = ?config(proxy_host, Config),
@@ -1046,12 +1045,19 @@ 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,
     Cluster = emqx_common_test_helpers:emqx_cluster(
         [core, core],
         [
             {apps, [emqx_conf, emqx_bridge, emqx_rule_engine]},
             {listener_ports, []},
-            {peer_mod, slave},
+            {peer_mod, PeerModule},
             {priv_data_dir, PrivDataDir},
             {load_schema, true},
             {start_autocluster, true},