Implement force_vhost_queues_shrink_member_to_current_member/1

(cherry picked from commit c26aa3b)
(cherry picked from commit b03637f)
(cherry picked from commit 023a46e)

Author: Ayanda-D

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -76,6 +76,7 @@
 -export([validate_policy/1, merge_policy_value/3]).
 
 -export([force_shrink_member_to_current_member/2,
+         force_vhost_queues_shrink_member_to_current_member/1,
          force_all_queues_shrink_member_to_current_member/0]).
 
 -ifdef(TEST).
@@ -1909,8 +1910,17 @@ force_shrink_member_to_current_member(VHost, Name) ->
             {error, not_found}
     end.
 
+force_vhost_queues_shrink_member_to_current_member(VHost) when is_binary(VHost) ->
+    rabbit_log:warning("Disaster recovery procedure: shrinking all quorum queues in vhost ~tp to a single node cluster", [VHost]),
+    ListQQs = fun() -> rabbit_amqqueue:list(VHost) end,
+    force_all_queues_shrink_member_to_current_member(ListQQs).
+
 force_all_queues_shrink_member_to_current_member() ->
     rabbit_log:warning("Disaster recovery procedure: shrinking all quorum queues to a single node cluster"),
+    ListQQs = fun() -> rabbit_amqqueue:list() end,
+    force_all_queues_shrink_member_to_current_member(ListQQs).
+
+force_all_queues_shrink_member_to_current_member(ListQQFun) when is_function(ListQQFun) ->
     Node = node(),
     _ = [begin
              QName = amqqueue:get_name(Q),
@@ -1925,7 +1935,7 @@ force_all_queues_shrink_member_to_current_member() ->
                    end,
              _ = rabbit_amqqueue:update(QName, Fun),
              _ = [ra:force_delete_server(?RA_SYSTEM, {RaName, N}) || N <- OtherNodes]
-         end || Q <- rabbit_amqqueue:list(), amqqueue:get_type(Q) == ?MODULE],
+         end || Q <- ListQQFun(), amqqueue:get_type(Q) == ?MODULE],
     rabbit_log:warning("Disaster recovery procedure: shrinking finished"),
     ok.