Fix more test flakes (QQs, MQTT, STOMP, Stream Protocol)

Now every updated test passes 10 times out of 10
locally.

(cherry picked from commit f364a7f)

# Conflicts:
#	deps/rabbitmq_stomp/test/connections_SUITE.erl
#	deps/rabbitmq_stomp/test/system_SUITE.erl

Author: michaelklishin

deps/rabbit/test/quorum_queue_member_reconciliation_SUITE.erl

@@ -12,6 +12,8 @@
 -include_lib("amqp_client/include/amqp_client.hrl").
 -compile([nowarn_export_all, export_all]).
 
+-import(rabbit_ct_helpers, [consistently/3]).
+
 %% The reconciler has two modes of triggering itself
 %% - timer based
 %% - event based
@@ -136,11 +138,11 @@ auto_grow(Config) ->
 
     add_server_to_cluster(Server0, Server1),
     %% With 2 nodes in the cluster, target group size is not reached, so no
-    %% new members should be available. We sleep a while so the periodic check
-    %% runs
-    timer:sleep(4000),
-    {ok, Members, _} = ra:members({queue_utils:ra_name(QQ), Server1}),
-    ?assertEqual(1, length(Members)),
+    %% new members should be available. Verify this holds over multiple
+    %% reconciliation cycles.
+    consistently(
+        ?_assertEqual(1, length(element(2, ra:members({queue_utils:ra_name(QQ), Server1})))),
+        1000, 4),
 
     add_server_to_cluster(Server2, Server1),
     %% With 3 nodes in the cluster, target size is met so eventually it should
@@ -175,10 +177,11 @@ auto_grow_drained_node(Config) ->
         fun () -> rabbit_ct_broker_helpers:is_being_drained_local_read(Config, Server0) end,
         10000),
     add_server_to_cluster(Server2, Server1),
-    timer:sleep(5000),
-    %% We have 3 nodes, but one is drained, so it will not be concidered.
-    {ok, Members1, _} = ra:members({queue_utils:ra_name(QQ), Server1}),
-    ?assertEqual(1, length(Members1)),
+    %% We have 3 nodes, but one is drained, so it will not be considered.
+    %% Verify this holds over multiple reconciliation cycles.
+    consistently(
+        ?_assertEqual(1, length(element(2, ra:members({queue_utils:ra_name(QQ), Server1})))),
+        1000, 5),
 
     rabbit_ct_broker_helpers:unmark_as_being_drained(Config, Server0),
     rabbit_ct_helpers:await_condition(

deps/rabbit/test/rabbit_fifo_int_SUITE.erl

@@ -534,9 +534,6 @@ handles_reject_notification(Config) ->
     % reverse order - should try the first node in the list first
     F0 = rabbit_fifo_client:init([ServerId2, ServerId1]),
     {ok, F1, []} = rabbit_fifo_client:enqueue(ClusterName, one, F0),
-
-    timer:sleep(500),
-
     % the applied notification
     _F2 = process_ra_events(receive_ra_events(1, 0), ClusterName, F1),
     rabbit_quorum_queue:stop_server(ServerId1),
@@ -646,10 +643,10 @@ cancel_checkout_with_pending(Config, Reason) ->
            end, F3, Msgs),
 
     {ok, _F4} = rabbit_fifo_client:cancel_checkout(<<"tag">>, Reason, F4),
-    timer:sleep(100),
-    {ok, Overview, _} = ra:member_overview(ServerId),
-    ?assertMatch(#{machine := #{num_messages := 0,
-                                num_consumers := 0}}, Overview),
+    rabbit_ct_helpers:eventually(
+        ?_assertMatch(#{machine := #{num_messages := 0, num_consumers := 0}},
+                      element(2, ra:member_overview(ServerId))),
+        200, 20),
     flush(),
     ok.
 

deps/rabbitmq_mqtt/test/auth_SUITE.erl

@@ -1031,7 +1031,8 @@ publish_permission_will_message(Config) ->
                            <<"client-with-will">>,
                            Opts),
     {ok, _} = emqtt:connect(C),
-    timer:sleep(100),
+    rabbit_ct_helpers:await_condition(
+      fun() -> util:all_connection_pids(Config) =/= [] end, 5_000),
     [ServerPublisherPid] = util:all_connection_pids(Config),
 
     Sub = open_mqtt_connection(Config),

deps/rabbitmq_mqtt/test/disconnect_on_unauthorized_SUITE.erl

@@ -88,9 +88,8 @@ publish_unauthorized_no_disconnect(Config) ->
 
     ok = emqtt:publish(C, <<"topic3">>, <<"payload">>, [{qos, 0}]),
 
-    timer:sleep(100),
     %% Client should be still connected.
-    ?assert(is_process_alive(C)),
+    rabbit_ct_helpers:consistently(?_assert(is_process_alive(C)), 50, 3),
 
     ok = emqtt:disconnect(C).
 
@@ -109,9 +108,8 @@ subscribe_unauthorized_no_disconnect(Config) ->
             ?assertEqual(?RC_FAILURE, ReasonCode)
     end,
 
-    timer:sleep(100),
     %% Client should be still connected.
-    ?assert(is_process_alive(C)),
+    rabbit_ct_helpers:consistently(?_assert(is_process_alive(C)), 50, 3),
 
     ok = emqtt:disconnect(C).
 
@@ -134,9 +132,8 @@ unsubscribe_unauthorized_no_disconnect(Config) ->
 
     {ok, _, ReasonCodes0} = emqtt:unsubscribe(C, <<"topic1">>),
 
-    timer:sleep(100),
     %% Client should be still connected.
-    ?assert(is_process_alive(C)),
+    rabbit_ct_helpers:consistently(?_assert(is_process_alive(C)), 50, 3),
 
     ok = rabbit_ct_broker_helpers:rpc(
            Config, rabbit_auth_backend_internal, set_topic_permissions,

deps/rabbitmq_mqtt/test/mqtt_shared_SUITE.erl

@@ -970,10 +970,7 @@ session_expiry(Config) ->
     ok = emqtt:disconnect(C),
 
     ?assertEqual(2, rpc(Config, rabbit_amqqueue, count, [])),
-    timer:sleep(timer:seconds(Seconds) + 100),
-    %% On a slow machine, this test might fail. Let's consider
-    %% the expiry on a longer time window
-    ?awaitMatch(0,  rpc(Config, rabbit_amqqueue, count, []), 15_000, 1000),
+    ?awaitMatch(0, rpc(Config, rabbit_amqqueue, count, []), 15_000, 1000),
 
     ok = rpc(Config, application, set_env, [App, Par, DefaultVal]).
 
@@ -1235,11 +1232,16 @@ rabbit_mqtt_qos0_queue_kill_node(Config) ->
     ok = emqtt:publish(Pub, Topic1, <<"m0">>, qos0),
     ok = expect_publishes(Sub0, Topic1, [<<"m0">>]),
 
+    [Node0 | _] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
     process_flag(trap_exit, true),
     ok = rabbit_ct_broker_helpers:kill_node(Config, 0),
     ok = await_exit(Sub0),
-    %% Wait to run rabbit_amqqueue:on_node_down/1 on both live nodes.
-    timer:sleep(500),
+    %% Wait for the live nodes to detect that node 0 has gone down,
+    %% otherwise the subsequent reconnect on node 1 may briefly stall
+    %% on cluster lookups that try to reach node 0.
+    rabbit_ct_helpers:await_condition(
+        fun() -> not lists:member(Node0, rpc(Config, 1, rabbit_nodes, list_running, [])) end,
+        10_000),
     %% Re-connect to a live node with same MQTT client ID.
     Sub1 = connect(SubscriberId, Config, 1, []),
     {ok, _, [0]} = emqtt:subscribe(Sub1, Topic2, qos0),
@@ -1390,12 +1392,15 @@ maintenance(Config) ->
     C1b = connect(<<"client-1b">>, Config, 1, []),
     ClientsNode1 = [C1a, C1b],
 
-    timer:sleep(500),
+    rabbit_ct_helpers:await_condition(
+        fun() -> length(all_connection_pids(Config)) =:= 3 end,
+        10_000),
 
     ok = drain_node(Config, 2),
     ok = revive_node(Config, 2),
-    timer:sleep(500),
-    [?assert(erlang:is_process_alive(C)) || C <- [C0, C1a, C1b]],
+    rabbit_ct_helpers:await_condition(
+        fun() -> lists:all(fun erlang:is_process_alive/1, [C0, C1a, C1b]) end,
+        10_000),
 
     process_flag(trap_exit, true),
     ok = drain_node(Config, 1),
@@ -1544,8 +1549,10 @@ block(Config) ->
     {ok, _} = emqtt:publish(C, Topic, <<"Not blocked yet">>, [{qos, 1}]),
 
     ok = rpc(Config, vm_memory_monitor, set_vm_memory_high_watermark, [0]),
-    %% Let it block
-    timer:sleep(100),
+    rabbit_ct_helpers:await_condition(fun() ->
+        Alarms = rpc(Config, rabbit_alarm, get_alarms, []),
+        lists:any(fun({{resource_limit, memory, _}, _}) -> true; (_) -> false end, Alarms)
+    end, 10_000),
 
     %% Blocked, but still will publish when unblocked
     puback_timeout = publish_qos1_timeout(C, Topic, <<"Now blocked">>, 1000),
@@ -1575,8 +1582,10 @@ block_only_publisher(Config) ->
     ok = expect_publishes(PubSub, Topic, [<<"from Pub">>, <<"from PubSub">>]),
 
     ok = rpc(Config, vm_memory_monitor, set_vm_memory_high_watermark, [0]),
-    %% Let it block
-    timer:sleep(100),
+    rabbit_ct_helpers:await_condition(fun() ->
+        Alarms = rpc(Config, rabbit_alarm, get_alarms, []),
+        lists:any(fun({{resource_limit, memory, _}, _}) -> true; (_) -> false end, Alarms)
+    end, 10_000),
 
     %% We expect that the publishing connections are blocked.
     [?assertEqual({error, ack_timeout}, emqtt:ping(Pid)) || Pid <- [Pub, PubSub]],
@@ -1594,8 +1603,10 @@ block_only_publisher(Config) ->
     ?assertEqual(pong, emqtt:ping(Sub)),
 
     rpc(Config, vm_memory_monitor, set_vm_memory_high_watermark, [0.6]),
-    %% Let it unblock
-    timer:sleep(100),
+    rabbit_ct_helpers:await_condition(fun() ->
+        Alarms = rpc(Config, rabbit_alarm, get_alarms, []),
+        not lists:any(fun({{resource_limit, memory, _}, _}) -> true; (_) -> false end, Alarms)
+    end, 10_000),
 
     %% All connections are unblocked.
     [?assertEqual(pong, emqtt:ping(Pid)) || Pid <- [Con, Sub, Pub, PubSub]],

deps/rabbitmq_mqtt/test/reader_SUITE.erl

@@ -102,8 +102,10 @@ block_connack_timeout(Config) ->
 
     DefaultWatermark = rpc(Config, vm_memory_monitor, get_vm_memory_high_watermark, []),
     ok = rpc(Config, vm_memory_monitor, set_vm_memory_high_watermark, [0]),
-    %% Let connection block.
-    timer:sleep(100),
+    rabbit_ct_helpers:await_condition(fun() ->
+        Alarms = rpc(Config, rabbit_alarm, get_alarms, []),
+        lists:any(fun({{resource_limit, memory, _}, _}) -> true; (_) -> false end, Alarms)
+    end, 10_000),
 
     %% We can still connect via TCP, but CONNECT packet will not be processed on the server.
     {ok, Client} = emqtt:start_link([{host, "localhost"},
@@ -162,11 +164,11 @@ login_timeout(Config) ->
 
 stats(Config) ->
     C = connect(?FUNCTION_NAME, Config),
-    %% Wait for stats being emitted (every 100ms)
-    timer:sleep(300),
-    %% Retrieve the connection Pid
     [Pid] = all_connection_pids(Config),
     [{pid, Pid}] = rpc(Config, rabbit_mqtt_reader, info, [Pid, [pid]]),
+    rabbit_ct_helpers:await_condition(fun() ->
+        rpc(Config, ets, lookup, [connection_metrics, Pid]) =/= []
+    end, 10_000),
     %% Verify the content of the metrics, garbage_collection must be present
     [{Pid, Props}] = rpc(Config, ets, lookup, [connection_metrics, Pid]),
     true = proplists:is_defined(garbage_collection, Props),

deps/rabbitmq_stomp/test/connections_SUITE.erl

@@ -91,7 +91,11 @@ direct_client_connections_are_not_leaked(Config) ->
                            Client, "LOL", [{"", ""}])
                   end,
                   lists:seq(1, 100)),
+<<<<<<< HEAD
     ?awaitMatch(N, count_connections(Config), 30_000),
+=======
+    rabbit_ct_helpers:await_condition(fun() -> count_connections(Config) =:= N end, 30_000),
+>>>>>>> f364a7fce (Fix more test flakes (QQs, MQTT, STOMP, Stream Protocol))
     ok.
 
 messages_not_dropped_on_disconnect(Config) ->

deps/rabbitmq_stomp/test/system_SUITE.erl

@@ -119,6 +119,70 @@ end_per_testcase0(publish_unauthorized_error, Config) ->
 end_per_testcase0(_, Config) ->
     Config.
 
+<<<<<<< HEAD
+=======
+transaction_limit(Config) ->
+    Client = ?config(stomp_client, Config),
+    %% Open 16 transactions (the limit)
+    lists:foreach(fun(I) ->
+        TxId = integer_to_binary(I),
+        rabbit_stomp_client:send(Client, 'BEGIN',
+            [{<<"transaction">>, TxId}])
+    end, lists:seq(1, 16)),
+
+    %% The 17th should fail
+    rabbit_stomp_client:send(Client, 'BEGIN',
+        [{<<"transaction">>, <<"17">>}]),
+    {ok, _Client1, Hdrs, _} = stomp_receive(Client, 'ERROR'),
+    <<"Transaction limit exceeded">> = maps:get(<<"message">>, Hdrs),
+    ok.
+
+global_counters(Config) ->
+    Version = ?config(version, Config),
+    ProtoVer = stomp_proto_ver(Version),
+    Dest = iolist_to_binary(["/topic/counter-test-", Version]),
+
+    C0 = get_global_counters(Config, ProtoVer),
+    Pubs0 = maps:get(publishers, C0, 0),
+    Cons0 = maps:get(consumers, C0, 0),
+    Recv0 = maps:get(messages_received_total, C0, 0),
+    Routed0 = maps:get(messages_routed_total, C0, 0),
+
+    Client = ?config(stomp_client, Config),
+    rabbit_stomp_client:send(
+      Client, 'SUBSCRIBE',
+      [{<<"destination">>, Dest}, {<<"id">>, <<"counter-sub">>}]),
+
+    rabbit_stomp_client:send(
+      Client, 'SEND', [{<<"destination">>, Dest}], ["hello"]),
+
+    {ok, Client1, _Hdrs, _Body} = stomp_receive(Client, 'MESSAGE'),
+
+    C1 = get_global_counters(Config, ProtoVer),
+    ?assertEqual(Pubs0 + 1, maps:get(publishers, C1)),
+    ?assertEqual(Cons0 + 1, maps:get(consumers, C1)),
+    ?assertEqual(Recv0 + 1, maps:get(messages_received_total, C1)),
+    ?assertEqual(Routed0 + 1, maps:get(messages_routed_total, C1)),
+
+    rabbit_stomp_client:send(
+      Client1, 'UNSUBSCRIBE', [{<<"id">>, <<"counter-sub">>}]),
+
+    rabbit_ct_helpers:await_condition(
+      fun() -> maps:get(consumers, get_global_counters(Config, ProtoVer)) =:= Cons0 end,
+      5_000),
+
+    ok.
+
+get_global_counters(Config, ProtoVer) ->
+    maps:get(#{protocol => ProtoVer},
+             rabbit_ct_broker_helpers:rpc(
+               Config, 0, rabbit_global_counters, overview, [])).
+
+stomp_proto_ver("1.0") -> 'STOMP 1.0';
+stomp_proto_ver("1.1") -> 'STOMP 1.1';
+stomp_proto_ver("1.2") -> 'STOMP 1.2'.
+
+>>>>>>> f364a7fce (Fix more test flakes (QQs, MQTT, STOMP, Stream Protocol))
 publish_no_dest_error(Config) ->
     Client = ?config(stomp_client, Config),
     rabbit_stomp_client:send(

deps/rabbitmq_stream/test/protocol_interop_SUITE.erl

@@ -60,8 +60,11 @@ init_per_testcase(Testcase, Config) ->
     rabbit_ct_helpers:testcase_started(Config, Testcase).
 
 end_per_testcase(Testcase, Config) ->
-    %% Wait for exclusive or auto-delete queues being deleted.
-    timer:sleep(800),
+    rabbit_ct_helpers:await_condition(
+        fun() ->
+            Qs = rabbit_ct_broker_helpers:rpc(Config, rabbit_amqqueue, list, []),
+            not lists:any(fun(Q) -> amqqueue:get_exclusive_owner(Q) =/= none end, Qs)
+        end, 10_000),
     rabbit_ct_broker_helpers:rpc(Config, ?MODULE, delete_queues, []),
     rabbit_ct_helpers:testcase_finished(Config, Testcase).