Merge pull request #14390 from thalesmg/20241211-r58-test-flaky-listener

test: fix flaky tests

apps/emqx/test/emqx_crl_cache_SUITE.erl

@@ -229,6 +229,18 @@ end_per_testcase(_TestCase, Config) ->
         proplists:get_value(tc_apps, Config)
     ),
     catch meck:unload([emqx_crl_cache]),
+    case whereis(emqx_crl_cache) of
+        Pid when is_pid(Pid) ->
+            MRef = monitor(process, Pid),
+            unlink(Pid),
+            exit(Pid, kill),
+            receive
+                {'DOWN', MRef, process, Pid, _} ->
+                    ok
+            end;
+        _ ->
+            ok
+    end,
     ok.
 
 %%--------------------------------------------------------------------
@@ -874,6 +886,7 @@ t_revoked(Config) ->
     ClientCert = filename:join(DataDir, "client-revoked.cert.pem"),
     ClientKey = filename:join(DataDir, "client-revoked.key.pem"),
     {ok, C} = emqtt:start_link([
+        {connect_timeout, 2},
         {ssl, true},
         {ssl_opts, [
             {certfile, ClientCert},

apps/emqx/test/emqx_listeners_SUITE.erl

@@ -272,11 +272,13 @@ t_ssl_update_opts(Config) ->
     Name = ?FUNCTION_NAME,
     with_listener(ssl, Name, Conf, fun() ->
         %% Client connects successfully.
+        ct:pal("attempting successful connection"),
         C1 = emqtt_connect_ssl(Host, Port, [
             {cacertfile, filename:join(PrivDir, "ca.pem")} | ClientSSLOpts
         ]),
 
         %% Change the listener SSL configuration: another set of cert/key files.
+        ct:pal("updating config"),
         {ok, _} = emqx:update_config(
             [listeners, ssl, Name],
             {update, #{
@@ -289,6 +291,7 @@ t_ssl_update_opts(Config) ->
         ),
 
         %% Unable to connect with old SSL options, server's cert is signed by another CA.
+        ct:pal("attempting connection with unknown CA"),
         ?assertError(
             {tls_alert, {unknown_ca, _}},
             emqtt_connect_ssl(Host, Port, [
@@ -312,13 +315,21 @@ t_ssl_update_opts(Config) ->
         ),
 
         %% Unable to connect with old SSL options, certificate is now required.
-        ?assertExceptionOneOf(
-            {error, {ssl_error, _Socket, {tls_alert, {certificate_required, _}}}},
-            {error, closed},
+        try
             emqtt_connect_ssl(Host, Port, [
                 {cacertfile, filename:join(PrivDir, "ca-next.pem")} | ClientSSLOpts
-            ])
-        ),
+            ]),
+            ct:fail("l ~b: unexpected success", [?LINE])
+        catch
+            error:{ssl_error, _Socket, {tls_alert, {certificate_required, _}}} ->
+                ok;
+            error:closed ->
+                ok;
+            error:connack_timeout ->
+                ok;
+            K:E:S ->
+                error({unexpected_exception, {K, E, S}})
+        end,
 
         C3 = emqtt_connect_ssl(Host, Port, [
             {cacertfile, filename:join(PrivDir, "ca-next.pem")},
@@ -359,6 +370,7 @@ t_wss_update_opts(Config) ->
     Name = ?FUNCTION_NAME,
     with_listener(wss, Name, Conf, fun() ->
         %% Start a client.
+        ct:pal("attempting successful connection"),
         C1 = emqtt_connect_wss(Host, Port, [
             {cacertfile, filename:join(PrivDir, "ca.pem")}
             | ClientSSLOpts
@@ -367,6 +379,7 @@ t_wss_update_opts(Config) ->
         %% Change the listener SSL configuration.
         %% 1. Another set of (password protected) cert/key files.
         %% 2. Require peer certificate.
+        ct:pal("changing config"),
         {ok, _} = emqx:update_config(
             [listeners, wss, Name],
             {update, #{
@@ -379,11 +392,16 @@ t_wss_update_opts(Config) ->
         ),
 
         %% Unable to connect with old SSL options, server's cert is signed by another CA.
+        ct:pal("attempting connection with unknown CA"),
         ?assertError(
             timeout,
-            emqtt_connect_wss(Host, Port, ClientSSLOpts)
+            emqtt_connect_wss(Host, Port, [
+                {cacerts, public_key:cacerts_get()}
+                | ClientSSLOpts
+            ])
         ),
 
+        ct:pal("attempting connection with another CA"),
         C2 = emqtt_connect_wss(Host, Port, [
             {cacertfile, filename:join(PrivDir, "ca-next.pem")}
             | ClientSSLOpts
@@ -656,7 +674,7 @@ with_listener(Type, Name, Config, Then) ->
 emqtt_connect_ssl(Host, Port, SSLOpts) ->
     emqtt_connect(fun emqtt:connect/1, #{
         hosts => [{Host, Port}],
-        connect_timeout => 1,
+        connect_timeout => 2,
         ssl => true,
         ssl_opts => SSLOpts
     }).
@@ -664,7 +682,7 @@ emqtt_connect_ssl(Host, Port, SSLOpts) ->
 emqtt_connect_quic(Host, Port, SSLOpts) ->
     emqtt_connect(fun emqtt:quic_connect/1, #{
         hosts => [{Host, Port}],
-        connect_timeout => 1,
+        connect_timeout => 2,
         ssl => true,
         ssl_opts => SSLOpts
     }).
@@ -672,7 +690,7 @@ emqtt_connect_quic(Host, Port, SSLOpts) ->
 emqtt_connect_wss(Host, Port, SSLOpts) ->
     emqtt_connect(fun emqtt:ws_connect/1, #{
         hosts => [{Host, Port}],
-        connect_timeout => 1,
+        connect_timeout => 2,
         ws_transport_options => [
             {protocols, [http]},
             {transport, tls},

apps/emqx_bridge_kafka/test/emqx_bridge_v2_kafka_consumer_SUITE.erl

@@ -560,13 +560,23 @@ t_pretty_api_dry_run_reason(Config) ->
                 ),
                 ?assertMatch({400, _}, Res),
                 {400, #{<<"message">> := Msg}} = Res,
-                ?assertEqual(
-                    match,
-                    re:run(Msg, <<"Leader for partition . unavailable; reason: ">>, [
-                        {capture, none}
-                    ]),
-                    #{message => Msg}
-                )
+                LeaderUnavailable =
+                    match ==
+                        re:run(
+                            Msg,
+                            <<"Leader for partition . unavailable; reason: ">>,
+                            [{capture, none}]
+                        ),
+                %% In CI, if this tests runs soon enough, Kafka may not be stable yet, and
+                %% this failure might occur.
+                CoordinatorFailure =
+                    match ==
+                        re:run(
+                            Msg,
+                            <<"shutdown,coordinator_failure">>,
+                            [{capture, none}]
+                        ),
+                ?assert(LeaderUnavailable or CoordinatorFailure, #{message => Msg})
             end),
             %% Wait for recovery; avoids affecting other test cases due to Kafka restabilizing...
             ?retry(

scripts/ct/run.sh

@@ -362,7 +362,7 @@ fi
 set +e
 
 if [ "$STOP" = 'yes' ]; then
-    $DC down --remove-orphans
+    $DC down -t 0 --remove-orphans
 elif [ "$ATTACH" = 'yes' ]; then
     docker exec -it "$ERLANG_CONTAINER" bash
 elif [ "$CONSOLE" = 'yes' ]; then