Shovel: de-flake dynamic_SUITE

checking that not a single process has a message in the mailbox
is prone to flakes.

deps/rabbitmq_shovel/test/dynamic_SUITE.erl

@@ -10,6 +10,8 @@
 -include_lib("eunit/include/eunit.hrl").
 -include_lib("amqp_client/include/amqp_client.hrl").
 
+-import(rabbit_ct_helpers, [eventually/1]).
+
 -compile(export_all).
 
 -export([spawn_suspender_proc/1]).
@@ -696,9 +698,11 @@ credit_flow(Config) ->
                     5000),
 
                   %% There should be only one process with a message buildup
-                  [{WriterPid, MQLen, _}, {_, 0, _} | _] =
+                  Top = [{WriterPid, MQLen, _}, {_, P, _} | _] =
                       rabbit_ct_broker_helpers:rpc(
                         Config, 0, recon, proc_count, [message_queue_len, 10]),
+                  ct:pal("Top processes by message queue length: ~p", [Top]),
+                  ?assert(P < 3),
 
                   %% The writer process should have only a limited
                   %% message queue. The shovel stops sending messages
@@ -725,9 +729,10 @@ credit_flow(Config) ->
                     end,
                     5000),
                   #{messages := 1000} = message_count(Config, <<"dest">>),
-                  [{_, 0, _}] =
+                  [{_, P, _}] =
                       rabbit_ct_broker_helpers:rpc(
                         Config, 0, recon, proc_count, [message_queue_len, 1]),
+                  ?assert(P < 3),
 
                   %% Status only transitions from flow to running
                   %% after a 1 second state-change-interval
@@ -839,9 +844,12 @@ dest_resource_alarm(AckMode, Config) ->
                   MsgCnts = message_count(Config, <<"src">>),
 
                   %% There should be no process with a message buildup
-                  [{_, 0, _}] =
-                      rabbit_ct_broker_helpers:rpc(
+                  eventually(?_assertEqual(0, begin
+                      Top = [{_, P, _}] = rabbit_ct_broker_helpers:rpc(
                           Config, 0, recon, proc_count, [message_queue_len, 1]),
+                      ct:pal("Top process by message queue length: ~p", [Top]),
+                      P
+                  end)),
 
                   %% Clear the resource alarm, all messages should
                   %% arrive to the dest queue