Closes #8564

rabbit_amqqueue:list* functions: take queue type into account when determining if a queue is local to a set of nodes.
rabbitmq · Jun 18, 2023 · 499a7c7 · 499a7c7
1 parent 8ad0e53
commit 499a7c7
Showing 1 changed file with 9 additions and 3 deletions.
diff --git a/deps/rabbit/src/rabbit_amqqueue.erl b/deps/rabbit/src/rabbit_amqqueue.erl
@@ -46,6 +46,7 @@
          list_local_mirrored_classic_without_synchronised_mirrors_for_cli/0,
          list_local_quorum_queues_with_name_matching/1,
          list_local_quorum_queues_with_name_matching/2]).
+-export([is_local_to_node/2, is_local_to_node_set/2]).
 -export([ensure_rabbit_queue_record_is_initialized/1]).
 -export([format/1]).
 -export([delete_immediately_by_resource/1]).
@@ -1096,7 +1097,7 @@ list() ->
     lists:filter(fun (Q) ->
                          Pid = amqqueue:get_pid(Q),
                          St = amqqueue:get_state(Q),
-                         St =/= stopped orelse lists:member(node(Pid), NodesRunning)
+                         St =/= stopped orelse is_local_to_node_set(Pid, NodesRunning)
                  end, All).
 
 -spec count() -> non_neg_integer().
@@ -1271,6 +1272,11 @@ is_local_to_node({_, Leader} = QPid, Node) when ?IS_QUORUM(QPid) ->
 is_local_to_node(_QPid, _Node) ->
     false.
 
+is_local_to_node_set(QPid, Nodes) when is_list(Nodes) ->
+  lists:any(fun(Node) -> is_local_to_node(QPid, Node) end, Nodes);
+is_local_to_node_set(QPid, OneNode) when is_atom(OneNode) ->
+  is_local_to_node(QPid, OneNode).
+
 is_in_virtual_host(Q, VHostName) ->
     VHostName =:= get_resource_vhost_name(amqqueue:get_name(Q)).
 
@@ -1281,7 +1287,7 @@ list(VHostPath) ->
     lists:filter(fun (Q) ->
                          Pid = amqqueue:get_pid(Q),
                          St = amqqueue:get_state(Q),
-                         St =/= stopped orelse lists:member(node(Pid), NodesRunning)
+                         St =/= stopped orelse is_local_to_node_set(Pid, NodesRunning)
                  end, All).
 
 -spec list_down(rabbit_types:vhost()) -> [amqqueue:amqqueue()].
@@ -1299,7 +1305,7 @@ list_down(VHostPath) ->
                       St = amqqueue:get_state(Q),
                       amqqueue:get_vhost(Q) =:= VHostPath
                           andalso
-                            ((St =:= stopped andalso not lists:member(node(Pid), NodesRunning))
+                            ((St =:= stopped andalso not is_local_to_node_set(Pid, NodesRunning))
                              orelse
                                (not sets:is_element(N, Alive)))
               end)