Apply PR suggestions

Author: LoisSotoLopez

deps/rabbit/src/rabbit_amqp_management.erl

@@ -471,7 +471,7 @@ encode_queue(Q, NumMsgs, NumConsumers) ->
     {Leader :: node() | none, Replicas :: [node(),...]}.
 queue_topology(Q) ->
     Leader = amqqueue:get_leader_node(Q),
-    Replicas = rabbit_amqqueue:get_nodes(Q),
+    Replicas = rabbit_queue_type:get_nodes(Q),
     {Leader, Replicas}.
 
 decode_exchange({map, KVList}) ->

deps/rabbit/src/rabbit_amqqueue.erl

@@ -23,7 +23,6 @@
 -export([list/0, list_durable/0, list/1, info_keys/0, info/1, info/2, info_all/1, info_all/2,
          emit_info_all/5, list_local/1, info_local/1,
          emit_info_local/4, emit_info_down/4]).
--export([get_nodes/1]).
 -export([count/0]).
 -export([list_down/1, list_down/2, list_all/1,
          count/1, list_names/0, list_names/1, list_local_names/0,
@@ -1232,12 +1231,6 @@ list() ->
 count() ->
     rabbit_db_queue:count().
 
--spec get_nodes(amqqueue:amqqueue_v2()) -> [node(),...].
-
-get_nodes(Q) ->
-    [{members, Nodes}] = info(Q, [members]),
-    Nodes.
-
 -spec list_names() -> [name()].
 
 list_names() ->
@@ -2051,7 +2044,7 @@ pseudo_queue(#resource{kind = queue} = QueueName, Pid, Durable)
                 ).
 
 get_quorum_nodes(Q) ->
-    rabbit_amqqueue:get_nodes(Q).
+    rabbit_queue_type:get_nodes(Q).
 
 -spec prepend_extra_bcc(Qs) ->
     Qs when Qs :: [amqqueue:target() | {amqqueue:target(), route_infos()}].

deps/rabbit/src/rabbit_queue_location.erl

@@ -143,7 +143,7 @@ select_members(Size, _, AllNodes, RunningNodes, _, _, GetQueues) ->
     Counters0 = maps:from_list([{N, 0} || N <- lists:delete(?MODULE:node(), AllNodes)]),
     Queues = GetQueues(),
     Counters = lists:foldl(fun(Q, Acc) ->
-                                   Nodes = rabbit_amqqueue:get_nodes(Q),
+                                   Nodes = rabbit_queue_type:get_nodes(Q),
                                    lists:foldl(fun(N, A)
                                                      when is_map_key(N, A) ->
                                                        maps:update_with(N, fun(C) -> C+1 end, A);

deps/rabbit/src/rabbit_queue_type.erl

@@ -38,6 +38,7 @@
          format/2,
          remove/2,
          info/2,
+         get_nodes/1,
          state_info/1,
          format_status/1,
          info_down/2,
@@ -413,6 +414,11 @@ info(Q, Items) ->
     Mod = amqqueue:get_type(Q),
     Mod:info(Q, Items).
 
+-spec get_nodes(amqqueue:amqqueue_v2()) -> [node(),...].
+get_nodes(Q) ->
+    [{members, Nodes}] = info(Q, [members]),
+    Nodes.
+
 fold_state(Fun, Acc, #?STATE{ctxs = Ctxs}) ->
     maps:fold(Fun, Acc, Ctxs).
 

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -272,13 +272,13 @@ start_cluster(Q) ->
                            || Node <- [LeaderNode | FollowerNodes]]),
     NewQ0 = amqqueue:set_pid(Q, LeaderId),
     NewQ1 = case rabbit_feature_flags:is_enabled(track_qq_members_uids) of
-        false ->
-            amqqueue:set_type_state(NewQ0,
-                                    #{nodes => [LeaderNode | FollowerNodes]});
-        true ->
-            amqqueue:set_type_state(NewQ0,
-                                    #{nodes => UIDs})
-    end,
+                false ->
+                    amqqueue:set_type_state(NewQ0,
+                                            #{nodes => [LeaderNode | FollowerNodes]});
+                true ->
+                    amqqueue:set_type_state(NewQ0,
+                                            #{nodes => UIDs})
+            end,
 
     Versions = [V || {ok, V} <- erpc:multicall(FollowerNodes,
                                                rabbit_fifo, version, [],
@@ -733,16 +733,14 @@ repair_amqqueue_nodes(Q0) ->
             %% update amqqueue record
             Fun = fun (Q) ->
                           TS0 = amqqueue:get_type_state(Q),
-                          TS = case rabbit_feature_flags:is_enabled(track_qq_members_uids)
-                                    andalso has_uuid_tracking(TS0)
-                               of
+                          TS = case rabbit_feature_flags:is_enabled(track_qq_members_uids) of
                                    false ->
                                        TS0#{nodes => RaNodes};
                                    true ->
                                        RaUids = maps:from_list([{N, erpc:call(N, ra_directory, uid_of,
-                                                               [?RA_SYSTEM, Name],
-                                                               ?RPC_TIMEOUT)}
-                                                 || N <- RaNodes]),
+                                                                              [?RA_SYSTEM, Name],
+                                                                              ?RPC_TIMEOUT)}
+                                                                || N <- RaNodes]),
                                        TS0#{nodes => RaUids}
                                end,
                           amqqueue:set_type_state(Q, TS)
@@ -803,28 +801,28 @@ maybe_apply_policies(Q, #{config := CurrentConfig}) ->
     {[amqqueue:amqqueue()], [amqqueue:amqqueue()]}.
 recover(_Vhost, Queues) ->
     lists:foldl(
-      fun (Q0, {R0, F0}) ->
-         {Name, _} = amqqueue:get_pid(Q0),
+      fun (Q, {R0, F0}) ->
+         {Name, _} = amqqueue:get_pid(Q),
          ServerId = {Name, node()},
-         QName = amqqueue:get_name(Q0),
-         MutConf = make_mutable_config(Q0),
+         QName = amqqueue:get_name(Q),
+         MutConf = make_mutable_config(Q),
          RaUId = ra_directory:uid_of(?RA_SYSTEM, Name),
-         #{nodes := Nodes} = QTypeState0 = amqqueue:get_type_state(Q0),
-         QTypeState = case Nodes of
+         #{nodes := Nodes} = amqqueue:get_type_state(Q),
+         case Nodes of
              List when is_list(List) ->
                  %% Queue is not aware of node to uid mapping, do nothing
-                 QTypeState0;
+                 ok;
              #{node() := RaUId} ->
-                 %% Queue is aware and uid for current node is correct, do nothing
-                 QTypeState0;
-             _ ->
-                 %% Queue is aware but either current node has no UId or it
-                 %% does not match the one returned by ra_directory, regen uid
-                 maybe_delete_data_dir(RaUId),
-                 NewRaUId = ra:new_uid(ra_lib:to_binary(Name)),
-                 QTypeState0#{nodes := Nodes#{node() => NewRaUId}}
+                 %% Queue is aware and uid for current node is correct, do
+                 %% nothing
+                 ok;
+             #{node() := _NewRaUId} ->
+                 %% Queue is aware but it does not match the one returned by
+                 %% ra_directory
+                 rabbit_log:info("Quorum queue ~ts: detected node uuid change, "
+                                 "deleting old data directory", [rabbit_misc:rs(QName)]),
+                 maybe_delete_data_dir(RaUId)
          end,
-         Q = amqqueue:set_type_state(Q0, QTypeState),
          Res = case ra:restart_server(?RA_SYSTEM, ServerId, MutConf) of
                    ok ->
                        % queue was restarted, good
@@ -1426,16 +1424,16 @@ do_add_member(Q0, Node, Membership, Timeout)
     QTypeState0 = #{nodes := Nodes} = amqqueue:get_type_state(Q0),
     NewRaUId = ra:new_uid(ra_lib:to_binary(RaName)),
     QTypeState = case Nodes of
-        L when is_list(L) ->
-            %% Queue is not aware of node to uid mapping, just add the new node
-            QTypeState0#{nodes => lists:usort([Node | Nodes])};
-        #{Node := _} ->
-            %% Queue is aware and uid for targeted node exists, do nothing
-            QTypeState0;
-        _ ->
-            %% Queue is aware but current node has no UId, regen uid
-            QTypeState0#{nodes => Nodes#{Node => NewRaUId}}
-    end,
+                     L when is_list(L) ->
+                         %% Queue is not aware of node to uid mapping, just add the new node
+                         QTypeState0#{nodes => lists:usort([Node | Nodes])};
+                     #{Node := _} ->
+                         %% Queue is aware and uid for targeted node exists, do nothing
+                         QTypeState0;
+                     _ ->
+                         %% Queue is aware but current node has no UId, regen uid
+                         QTypeState0#{nodes => Nodes#{Node => NewRaUId}}
+                 end,
     Q = amqqueue:set_type_state(Q0, QTypeState),
     MachineVersion = erpc_call(Node, rabbit_fifo, version, [], infinity),
     Conf = make_ra_conf(Q, ServerId, Membership, MachineVersion),
@@ -2049,13 +2047,13 @@ make_ra_conf(Q, ServerId, TickTimeout,
     [{ClusterName, _} | _] = Members = members(Q),
     {_, Node} = ServerId,
     UId = case amqqueue:get_type_state(Q) of
-        #{nodes := #{Node := Id}} ->
-            Id;
-        _ ->
-            %% Queue was declared on an older version of RabbitMQ
-            %% and does not have the node to uid mappings
-            ra:new_uid(ra_lib:to_binary(ClusterName))
-    end,
+              #{nodes := #{Node := Id}} ->
+                  Id;
+              _ ->
+                  %% Queue was declared on an older version of RabbitMQ
+                  %% or does not have the node to uid mappings
+                  ra:new_uid(ra_lib:to_binary(ClusterName))
+          end,
     FName = rabbit_misc:rs(QName),
     Formatter = {?MODULE, format_ra_event, [QName]},
     LogCfg = #{uid => UId,
@@ -2199,7 +2197,7 @@ force_checkpoint_on_queue(QName) ->
         {ok, Q} when ?amqqueue_is_quorum(Q) ->
             {RaName, _} = amqqueue:get_pid(Q),
             ?LOG_DEBUG("Sending command to force ~ts to take a checkpoint", [QNameFmt]),
-            Nodes = rabbit_amqqueue:get_nodes(Q),
+            Nodes = rabbit_queue_type:get_nodes(Q),
             _ = [ra:cast_aux_command({RaName, Node}, force_checkpoint)
                  || Node <- Nodes],
             ok;
@@ -2472,8 +2470,3 @@ queue_vm_ets() ->
 
 tick_interval() ->
     application:get_env(rabbit, quorum_tick_interval, ?TICK_INTERVAL).
-
-has_uuid_tracking(#{nodes := Nodes}) when is_map(Nodes) ->
-    true;
-has_uuid_tracking(_QTypeState) ->
-    false.

deps/rabbit/src/rabbit_stream_coordinator.erl

@@ -154,7 +154,7 @@ stop() ->
 new_stream(Q, LeaderNode)
   when ?is_amqqueue(Q) andalso is_atom(LeaderNode) ->
     #{name := StreamId} = amqqueue:get_type_state(Q),
-    Nodes = rabbit_amqqueue:get_nodes(Q),
+    Nodes = rabbit_queue_type:get_nodes(Q),
     %% assertion leader is in nodes configuration
     true = lists:member(LeaderNode, Nodes),
     process_command({new_stream, StreamId,

deps/rabbitmq_ct_helpers/src/queue_utils.erl

@@ -208,7 +208,7 @@ assert_number_of_replicas(Config, Server, VHost, QQ, Count) ->
        begin
            {ok, Q} = rabbit_ct_broker_helpers:rpc(
                        Config, Server, rabbit_amqqueue, lookup, [QQ, VHost]),
-           Nodes = rabbit_amqqueue:get_nodes(Q),
+           Nodes = rabbit_queue_type:get_nodes(Q),
            length(Nodes)
        end,
        30000).

deps/rabbitmq_prometheus/src/collectors/prometheus_rabbitmq_core_metrics_collector.erl

@@ -451,7 +451,7 @@ emit_queue_info(Prefix, VHostsFilter, Callback) ->
                            true -> Acc;
                            false ->
                                Type = amqqueue:get_type(Q),
-                               Members = rabbit_amqqueue:get_nodes(Q),
+                               Members = rabbit_queue_type:get_nodes(Q),
                                case membership(amqqueue:get_pid(Q), Members) of
                                    not_a_member ->
                                        Acc;