Merge pull request #9438 from rabbitmq/test-resilience

Author: michaelklishin

deps/rabbit/app.bzl

@@ -1025,7 +1025,7 @@ def test_suite_beam_files(name = "test_suite_beam_files"):
         outs = ["test/logging_SUITE.beam"],
         app_name = "rabbit",
         erlc_opts = "//:test_erlc_opts",
-        deps = ["//deps/amqp_client:erlang_app", "//deps/rabbit_common:erlang_app"],
+        deps = ["//deps/amqp_client:erlang_app", "//deps/rabbit_common:erlang_app", "//deps/rabbitmq_ct_helpers:erlang_app"],
     )
     erlang_bytecode(
         name = "lqueue_SUITE_beam_files",

deps/rabbit/test/logging_SUITE.erl

@@ -13,6 +13,7 @@
 -include_lib("kernel/include/logger.hrl").
 -include_lib("rabbit_common/include/logging.hrl").
 -include_lib("amqp_client/include/amqp_client.hrl").
+-include_lib("rabbitmq_ct_helpers/include/rabbit_assert.hrl").
 
 -export([suite/0,
          all/0,
@@ -922,18 +923,10 @@ logging_to_exchange_works(Config) ->
       #{exchange := #resource{name = XName} = Exchange}} = ExchangeHandler,
 
     %% Wait for the expected exchange to be automatically declared.
-    ?assert(
-       lists:any(
-         fun(_) ->
-                 Ret = rabbit_ct_broker_helpers:rpc(
-                         Config, 0,
-                         rabbit_exchange, lookup, [Exchange]),
-                 case Ret of
-                     {ok, _} -> true;
-                     _       -> timer:sleep(500),
-                                false
-                 end
-         end, lists:seq(1, 20))),
+    ?awaitMatch({ok, _}, rabbit_ct_broker_helpers:rpc(
+                           Config, 0,
+                           rabbit_exchange, lookup, [Exchange]),
+                10000),
 
     %% Declare a queue to collect all logged messages.
     {Conn, Chan} = rabbit_ct_client_helpers:open_connection_and_channel(

deps/rabbit/test/rabbit_fifo_dlx_integration_SUITE.erl

@@ -20,6 +20,7 @@
 
 -import(queue_utils, [wait_for_messages_ready/3,
                       wait_for_min_messages/3,
+                      wait_for_messages/2,
                       dirty_query/3,
                       ra_name/1]).
 -import(rabbit_ct_helpers, [eventually/1,
@@ -639,9 +640,10 @@ reject_publish_max_length_target_quorum_queue(Config) ->
     %% Make space in target queue by consuming messages one by one
     %% allowing for more dead-lettered messages to reach the target queue.
     [begin
-         timer:sleep(2000),
          Msg = integer_to_binary(N),
-         {#'basic.get_ok'{}, #amqp_msg{payload = Msg}} = amqp_channel:call(Ch, #'basic.get'{queue = TargetQ})
+         ?awaitMatch({#'basic.get_ok'{}, #amqp_msg{payload = Msg}},
+                     amqp_channel:call(Ch, #'basic.get'{queue = TargetQ}),
+                     ?DEFAULT_WAIT, ?DEFAULT_INTERVAL)
      end || N <- lists:seq(1,4)],
     eventually(?_assertEqual([{0, 0}],
                              dirty_query([Server], RaName, fun rabbit_fifo:query_stat_dlx/1)), 500, 10),
@@ -686,7 +688,7 @@ reject_publish_down_target_quorum_queue(Config) ->
      end || N <- lists:seq(21, 50)],
 
     %% The target queue should have all 50 messages.
-    timer:sleep(2000),
+    wait_for_messages(Config, [[TargetQ, <<"50">>, <<"50">>, <<"0">>]]),
     Received = lists:foldl(
                  fun(_N, S) ->
                          {#'basic.get_ok'{}, #amqp_msg{payload = Msg}} =