Rename quorum_queue_utils to queue_utils (backport #9417)

deps/rabbit/BUILD.bazel

@@ -363,7 +363,7 @@ rabbitmq_integration_suite(
     name = "consumer_timeout_SUITE",
     size = "medium",
     additional_beam = [
-        ":test_quorum_queue_utils_beam",
+        ":test_queue_utils_beam",
     ],
 )
 
@@ -376,7 +376,7 @@ rabbitmq_integration_suite(
     name = "dead_lettering_SUITE",
     size = "large",
     additional_beam = [
-        ":test_quorum_queue_utils_beam",
+        ":test_queue_utils_beam",
     ],
     shard_count = 7,
 )
@@ -411,7 +411,7 @@ rabbitmq_integration_suite(
     name = "dynamic_qq_SUITE",
     size = "large",
     additional_beam = [
-        ":test_quorum_queue_utils_beam",
+        ":test_queue_utils_beam",
     ],
     deps = [
         "@proper//:erlang_app",
@@ -493,7 +493,7 @@ rabbitmq_integration_suite(
     name = "maintenance_mode_SUITE",
     size = "medium",
     additional_beam = [
-        ":test_quorum_queue_utils_beam",
+        ":test_queue_utils_beam",
     ],
 )
 
@@ -624,7 +624,7 @@ rabbitmq_integration_suite(
     name = "publisher_confirms_parallel_SUITE",
     size = "medium",
     additional_beam = [
-        ":test_quorum_queue_utils_beam",
+        ":test_queue_utils_beam",
     ],
 )
 
@@ -643,7 +643,7 @@ rabbitmq_integration_suite(
     name = "queue_parallel_SUITE",
     size = "large",
     additional_beam = [
-        ":test_quorum_queue_utils_beam",
+        ":test_queue_utils_beam",
     ],
     shard_count = 6,
 )
@@ -657,8 +657,8 @@ rabbitmq_integration_suite(
     name = "quorum_queue_SUITE",
     size = "large",
     additional_beam = [
-        ":test_quorum_queue_utils_beam",
         ":test_clustering_utils_beam",
+        ":test_queue_utils_beam",
     ],
     shard_count = 6,
 )
@@ -718,7 +718,7 @@ rabbitmq_integration_suite(
     size = "medium",
     additional_beam = [
         ":test_test_util_beam",
-        ":test_quorum_queue_utils_beam",
+        ":test_queue_utils_beam",
         ":quorum_queue_SUITE_beam_files",
     ],
     deps = [
@@ -784,7 +784,7 @@ rabbitmq_integration_suite(
     name = "rabbit_stream_queue_SUITE",
     size = "large",
     additional_beam = [
-        ":test_quorum_queue_utils_beam",
+        ":test_queue_utils_beam",
     ],
     shard_count = 12,
     deps = [
@@ -826,7 +826,7 @@ rabbitmq_integration_suite(
     name = "single_active_consumer_SUITE",
     size = "medium",
     additional_beam = [
-        ":test_quorum_queue_utils_beam",
+        ":test_queue_utils_beam",
     ],
 )
 
@@ -1050,7 +1050,7 @@ rabbitmq_integration_suite(
     name = "direct_exchange_routing_v2_SUITE",
     size = "medium",
     additional_beam = [
-        ":test_quorum_queue_utils_beam",
+        ":test_queue_utils_beam",
     ],
 )
 
@@ -1071,7 +1071,7 @@ rabbitmq_integration_suite(
     name = "exchanges_SUITE",
     size = "medium",
     additional_beam = [
-        ":test_quorum_queue_utils_beam",
+        ":test_queue_utils_beam",
     ],
 )
 
@@ -1218,7 +1218,7 @@ eunit(
         ":test_dummy_supervisor2_beam",
         ":test_failing_dummy_interceptor_beam",
         ":test_mirrored_supervisor_SUITE_gs_beam",
-        ":test_quorum_queue_utils_beam",
+        ":test_queue_utils_beam",
         ":test_rabbit_auth_backend_context_propagation_mock_beam",
         ":test_rabbit_dummy_protocol_connection_info_beam",
         ":test_rabbit_foo_protocol_connection_info_beam",

deps/rabbit/app.bzl

@@ -1493,10 +1493,10 @@ def test_suite_beam_files(name = "test_suite_beam_files"):
         erlc_opts = "//:test_erlc_opts",
     )
     erlang_bytecode(
-        name = "test_quorum_queue_utils_beam",
+        name = "test_queue_utils_beam",
         testonly = True,
-        srcs = ["test/quorum_queue_utils.erl"],
-        outs = ["test/quorum_queue_utils.beam"],
+        srcs = ["test/queue_utils.erl"],
+        outs = ["test/queue_utils.beam"],
         app_name = "rabbit",
         erlc_opts = "//:test_erlc_opts",
     )

deps/rabbit/test/consumer_timeout_SUITE.erl

@@ -27,7 +27,7 @@
                                               {queue_policy, []},
                                               {queue_arguments, [{<<"x-consumer-timeout">>, long, ?CONSUMER_TIMEOUT}]}]}).
 
--import(quorum_queue_utils, [wait_for_messages/2]).
+-import(queue_utils, [wait_for_messages/2]).
 
 all() ->
     [

deps/rabbit/test/dead_lettering_SUITE.erl

@@ -15,7 +15,7 @@
 
 -compile([export_all, nowarn_export_all]).
 
--import(quorum_queue_utils, [wait_for_messages/2]).
+-import(queue_utils, [wait_for_messages/2]).
 
 all() ->
     [

deps/rabbit/test/direct_exchange_routing_v2_SUITE.erl

@@ -304,8 +304,8 @@ route_exchange_to_exchange(Config) ->
     bind_queue(Ch, Q2, FanoutX, <<"ignored">>),
 
     publish(Ch, DirectX, RKey),
-    quorum_queue_utils:wait_for_messages(Config, [[Q1, <<"1">>, <<"1">>, <<"0">>]]),
-    quorum_queue_utils:wait_for_messages(Config, [[Q2, <<"1">>, <<"1">>, <<"0">>]]),
+    queue_utils:wait_for_messages(Config, [[Q1, <<"1">>, <<"1">>, <<"0">>]]),
+    queue_utils:wait_for_messages(Config, [[Q2, <<"1">>, <<"1">>, <<"0">>]]),
     ?assertEqual(1, table_size(Config, ?INDEX_TABLE_NAME)),
 
     %% cleanup

deps/rabbit/test/dynamic_qq_SUITE.erl

@@ -11,8 +11,8 @@
 -include_lib("amqp_client/include/amqp_client.hrl").
 -include_lib("rabbitmq_ct_helpers/include/rabbit_assert.hrl").
 
--import(quorum_queue_utils, [wait_for_messages_ready/3,
-                             ra_name/1]).
+-import(queue_utils, [wait_for_messages_ready/3,
+                      ra_name/1]).
 
 -compile(nowarn_export_all).
 -compile(export_all).

deps/rabbit/test/exchanges_SUITE.erl

@@ -104,7 +104,7 @@ direct_exchange(Config) ->
     publish(Ch, Direct, Q, <<"msg1">>),
     publish(Ch, Direct, <<"anyotherkey">>, <<"msg2">>),
 
-    quorum_queue_utils:wait_for_messages(Config, [[Q, <<"1">>, <<"1">>, <<"0">>],
+    queue_utils:wait_for_messages(Config, [[Q, <<"1">>, <<"1">>, <<"0">>],
                                                   [AltQ, <<"0">>, <<"0">>, <<"0">>]]),
     ?assertMatch({#'basic.get_ok'{}, #amqp_msg{payload = <<"msg1">>}},
                  amqp_channel:call(Ch, #'basic.get'{queue = Q})),
@@ -139,7 +139,7 @@ topic_exchange(Config) ->
     publish(Ch, Topic, <<"simplycarrots">>, <<"msg5">>),
     publish(Ch, Topic, <<"*.queue.rules">>, <<"msg6">>),
 
-    quorum_queue_utils:wait_for_messages(Config, [[Q, <<"3">>, <<"3">>, <<"0">>]]),
+    queue_utils:wait_for_messages(Config, [[Q, <<"3">>, <<"3">>, <<"0">>]]),
     ?assertMatch({#'basic.get_ok'{}, #amqp_msg{payload = <<"msg1">>}},
                  amqp_channel:call(Ch, #'basic.get'{queue = Q})),
     ?assertMatch({#'basic.get_ok'{}, #amqp_msg{payload = <<"msg2">>}},
@@ -153,11 +153,11 @@ topic_exchange(Config) ->
                                                              queue = Q,
                                                              routing_key = <<"#.noclue">>}),
     publish(Ch, Topic, <<"simplycarrots">>, <<"msg7">>),
-    quorum_queue_utils:wait_for_messages(Config, [[Q, <<"3">>, <<"0">>, <<"3">>]]),
+    queue_utils:wait_for_messages(Config, [[Q, <<"3">>, <<"0">>, <<"3">>]]),
     publish(Ch, Topic, <<"#.bla">>, <<"msg8">>),
-    quorum_queue_utils:wait_for_messages(Config, [[Q, <<"3">>, <<"0">>, <<"3">>]]),
+    queue_utils:wait_for_messages(Config, [[Q, <<"3">>, <<"0">>, <<"3">>]]),
     publish(Ch, Topic, <<"#.noclue">>, <<"msg9">>),
-    quorum_queue_utils:wait_for_messages(Config, [[Q, <<"4">>, <<"1">>, <<"3">>]]),
+    queue_utils:wait_for_messages(Config, [[Q, <<"4">>, <<"1">>, <<"3">>]]),
     ?assertMatch({#'basic.get_ok'{}, #amqp_msg{payload = <<"msg9">>}},
                  amqp_channel:call(Ch, #'basic.get'{queue = Q})),
     ?assertMatch(#'basic.get_empty'{},
@@ -167,7 +167,7 @@ topic_exchange(Config) ->
                                                              queue = Q,
                                                              routing_key = <<"#">>}),
     publish(Ch, Topic, <<"simplycarrots">>, <<"msg10">>),
-    quorum_queue_utils:wait_for_messages(Config, [[Q, <<"5">>, <<"1">>, <<"4">>]]),
+    queue_utils:wait_for_messages(Config, [[Q, <<"5">>, <<"1">>, <<"4">>]]),
     ?assertMatch({#'basic.get_ok'{}, #amqp_msg{payload = <<"msg10">>}},
                  amqp_channel:call(Ch, #'basic.get'{queue = Q})),
     ?assertMatch(#'basic.get_empty'{},
@@ -193,7 +193,7 @@ fanout_exchange(Config) ->
     publish(Ch, Fanout, Q, <<"msg1">>),
     publish(Ch, Fanout, <<"anyotherkey">>, <<"msg2">>),
 
-    quorum_queue_utils:wait_for_messages(Config, [[Q, <<"2">>, <<"2">>, <<"0">>],
+    queue_utils:wait_for_messages(Config, [[Q, <<"2">>, <<"2">>, <<"0">>],
                                                   [AltQ, <<"2">>, <<"2">>, <<"0">>]]),
     ?assertMatch({#'basic.get_ok'{}, #amqp_msg{payload = <<"msg1">>}},
                  amqp_channel:call(Ch, #'basic.get'{queue = Q})),
@@ -241,7 +241,7 @@ headers_exchange(Config) ->
     publish(Ch, Headers, <<>>, <<"msg2">>, [{<<"foo">>, longstr, <<"bar">>}]),
     publish(Ch, Headers, <<>>, <<"msg3">>),
 
-    quorum_queue_utils:wait_for_messages(Config, [[Q, <<"1">>, <<"1">>, <<"0">>],
+    queue_utils:wait_for_messages(Config, [[Q, <<"1">>, <<"1">>, <<"0">>],
                                                   [AltQ, <<"2">>, <<"2">>, <<"0">>]]),
 
     ?assertMatch({#'basic.get_ok'{}, #amqp_msg{payload = <<"msg1">>}},

deps/rabbit/test/maintenance_mode_SUITE.erl

@@ -242,7 +242,7 @@ quorum_queue_leadership_transfer(Config) ->
         fun () -> rabbit_ct_broker_helpers:is_being_drained_local_read(Config, A) end, 10000),
 
     %% quorum queue leader election is asynchronous
-    AllTheSame = quorum_queue_utils:fifo_machines_use_same_version(
+    AllTheSame = queue_utils:fifo_machines_use_same_version(
                    Config, Nodenames),
     case AllTheSame of
         true ->

deps/rabbit/test/publisher_confirms_parallel_SUITE.erl

@@ -15,7 +15,7 @@
 
 -define(TIMEOUT, 60000).
 
--import(quorum_queue_utils, [wait_for_messages/2]).
+-import(queue_utils, [wait_for_messages/2]).
 
 all() ->
     [

deps/rabbit/test/quorum_queue_utils.erl → deps/rabbit/test/queue_utils.erl

@@ -1,4 +1,4 @@
--module(quorum_queue_utils).
+-module(queue_utils).
 
 -include_lib("eunit/include/eunit.hrl").
 

deps/rabbit/test/quorum_queue_SUITE.erl

@@ -11,7 +11,7 @@
 -include_lib("amqp_client/include/amqp_client.hrl").
 -include_lib("rabbitmq_ct_helpers/include/rabbit_assert.hrl").
 
--import(quorum_queue_utils, [wait_for_messages_ready/3,
+-import(queue_utils, [wait_for_messages_ready/3,
                              wait_for_messages_pending_ack/3,
                              wait_for_messages_total/3,
                              wait_for_messages/2,

deps/rabbit/test/rabbit_fifo_dlx_integration_SUITE.erl

@@ -18,10 +18,10 @@
 -include_lib("amqp_client/include/amqp_client.hrl").
 -include_lib("rabbitmq_ct_helpers/include/rabbit_assert.hrl").
 
--import(quorum_queue_utils, [wait_for_messages_ready/3,
-                             wait_for_min_messages/3,
-                             dirty_query/3,
-                             ra_name/1]).
+-import(queue_utils, [wait_for_messages_ready/3,
+                      wait_for_min_messages/3,
+                      dirty_query/3,
+                      ra_name/1]).
 -import(rabbit_ct_helpers, [eventually/1,
                             eventually/3,
                             consistently/1]).

deps/rabbit/test/rabbit_stream_queue_SUITE.erl

@@ -736,7 +736,7 @@ publish(Config) ->
                  declare(Ch, Q, [{<<"x-queue-type">>, longstr, <<"stream">>}])),
 
     publish(Ch, Q),
-    quorum_queue_utils:wait_for_messages(Config, [[Q, <<"1">>, <<"1">>, <<"0">>]]),
+    queue_utils:wait_for_messages(Config, [[Q, <<"1">>, <<"1">>, <<"0">>]]),
     rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, delete_testcase_queue, [Q]).
 
 publish_confirm(Config) ->
@@ -751,7 +751,7 @@ publish_confirm(Config) ->
     amqp_channel:register_confirm_handler(Ch, self()),
     publish(Ch, Q),
     amqp_channel:wait_for_confirms(Ch, 5),
-    quorum_queue_utils:wait_for_messages(Config, [[Q, <<"1">>, <<"1">>, <<"0">>]]),
+    queue_utils:wait_for_messages(Config, [[Q, <<"1">>, <<"1">>, <<"0">>]]),
     rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, delete_testcase_queue, [Q]).
 
 restart_single_node(Config) ->
@@ -762,15 +762,15 @@ restart_single_node(Config) ->
     ?assertEqual({'queue.declare_ok', Q, 0, 0},
                  declare(Ch, Q, [{<<"x-queue-type">>, longstr, <<"stream">>}])),
     publish(Ch, Q),
-    quorum_queue_utils:wait_for_messages(Config, [[Q, <<"1">>, <<"1">>, <<"0">>]]),
+    queue_utils:wait_for_messages(Config, [[Q, <<"1">>, <<"1">>, <<"0">>]]),
 
     rabbit_control_helper:command(stop_app, Server),
     rabbit_control_helper:command(start_app, Server),
 
-    quorum_queue_utils:wait_for_messages(Config, [[Q, <<"1">>, <<"1">>, <<"0">>]]),
+    queue_utils:wait_for_messages(Config, [[Q, <<"1">>, <<"1">>, <<"0">>]]),
     Ch1 = rabbit_ct_client_helpers:open_channel(Config, Server),
     publish(Ch1, Q),
-    quorum_queue_utils:wait_for_messages(Config, [[Q, <<"2">>, <<"2">>, <<"0">>]]),
+    queue_utils:wait_for_messages(Config, [[Q, <<"2">>, <<"2">>, <<"0">>]]),
     rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, delete_testcase_queue, [Q]).
 
 %% the failing case for this test relies on a particular random condition
@@ -816,7 +816,7 @@ recover(Config) ->
     ?assertEqual({'queue.declare_ok', Q, 0, 0},
                  declare(Ch, Q, [{<<"x-queue-type">>, longstr, <<"stream">>}])),
     publish(Ch, Q),
-    quorum_queue_utils:wait_for_messages(Config, [[Q, <<"1">>, <<"1">>, <<"0">>]]),
+    queue_utils:wait_for_messages(Config, [[Q, <<"1">>, <<"1">>, <<"0">>]]),
 
     Perm0 = permute(Servers0),
     Servers = lists:nth(rand:uniform(length(Perm0)), Perm0),
@@ -828,7 +828,7 @@ recover(Config) ->
     [rabbit_ct_broker_helpers:start_node(Config, S) || S <- lists:reverse(Servers)],
     ct:pal("recover: running stop waiting for messages ~w", [Servers]),
     check_leader_and_replicas(Config, Servers0),
-    quorum_queue_utils:wait_for_messages(Config, [[Q, <<"1">>, <<"1">>, <<"0">>]], 60),
+    queue_utils:wait_for_messages(Config, [[Q, <<"1">>, <<"1">>, <<"0">>]], 60),
 
     %% Another single random permutation
     Perm1 = permute(Servers0),
@@ -839,11 +839,11 @@ recover(Config) ->
     [rabbit_control_helper:command(start_app, S) || S <- lists:reverse(Servers1)],
     ct:pal("recover: running app stop waiting for messages ~w", [Servers1]),
     check_leader_and_replicas(Config, Servers0),
-    quorum_queue_utils:wait_for_messages(Config, [[Q, <<"1">>, <<"1">>, <<"0">>]], 60),
+    queue_utils:wait_for_messages(Config, [[Q, <<"1">>, <<"1">>, <<"0">>]], 60),
 
     Ch1 = rabbit_ct_client_helpers:open_channel(Config, Server),
     publish(Ch1, Q),
-    quorum_queue_utils:wait_for_messages(Config, [[Q, <<"2">>, <<"2">>, <<"0">>]]),
+    queue_utils:wait_for_messages(Config, [[Q, <<"2">>, <<"2">>, <<"0">>]]),
     rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, delete_testcase_queue, [Q]).
 
 restart_coordinator_without_queues(Config) ->
@@ -1356,7 +1356,7 @@ consume_and_ack(Config) ->
             %%
             ?assertMatch({'queue.declare_ok', Q, _MsgCount, 0},
                          declare(Ch1, Q, [{<<"x-queue-type">>, longstr, <<"stream">>}])),
-            quorum_queue_utils:wait_for_messages(Config, [[Q, <<"1">>, <<"1">>, <<"0">>]])
+            queue_utils:wait_for_messages(Config, [[Q, <<"1">>, <<"1">>, <<"0">>]])
     after 5000 ->
             exit(timeout)
     end,
@@ -1795,8 +1795,8 @@ max_age(Config) ->
     amqp_channel:wait_for_confirms(Ch, 5),
 
     %% Let's give it some margin if some messages fall between segments
-    quorum_queue_utils:wait_for_min_messages(Config, Q, 100),
-    quorum_queue_utils:wait_for_max_messages(Config, Q, 150),
+    queue_utils:wait_for_min_messages(Config, Q, 100),
+    queue_utils:wait_for_max_messages(Config, Q, 150),
 
     rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, delete_testcase_queue, [Q]).
 
@@ -2269,7 +2269,7 @@ update_retention_policy(Config) ->
 
     %% Retention policy should clear approximately 2/3 of the messages, but just to be safe
     %% let's simply check that it removed half of them
-    quorum_queue_utils:wait_for_max_messages(Config, Q, 5000),
+    queue_utils:wait_for_max_messages(Config, Q, 5000),
 
     {ok, Q1} = rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_amqqueue, lookup,
                                             [rabbit_misc:r(<<"/">>, queue, Q)]),
@@ -2389,7 +2389,7 @@ dead_letter_target(Config) ->
             ok = amqp_channel:cast(Ch1, #'basic.nack'{delivery_tag = DeliveryTag,
                                                       requeue =false,
                                                       multiple     = false}),
-            quorum_queue_utils:wait_for_messages(Config, [[Q, <<"1">>, <<"1">>, <<"0">>]])
+            queue_utils:wait_for_messages(Config, [[Q, <<"1">>, <<"1">>, <<"0">>]])
     after 5000 ->
             exit(timeout)
     end,