Shutdown peer QQ FSMs on connected nodes on force-shrink execution for cluster

wide consistency, ensuring only the leader is active/running

(cherry picked from commit b675ce2)

Author: Ayanda-D

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -1376,6 +1376,7 @@ delete_member(Q, Node) when ?amqqueue_is_quorum(Q) ->
                     _ = rabbit_amqqueue:update(QName, Fun),
                     case ra:force_delete_server(?RA_SYSTEM, ServerId) of
                         ok ->
+                            rabbit_log:info("Deleted a replica of quorum ~ts on node ~ts", [rabbit_misc:rs(QName), Node]),
                             ok;
                         {error, {badrpc, nodedown}} ->
                             ok;
@@ -1957,13 +1958,15 @@ force_shrink_member_to_current_member(VHost, Name) ->
     case rabbit_amqqueue:lookup(QName) of
         {ok, Q} when ?is_amqqueue(Q) ->
             {RaName, _} = amqqueue:get_pid(Q),
+            OtherNodes = lists:delete(Node, get_nodes(Q)),
             ok = ra_server_proc:force_shrink_members_to_current_member({RaName, Node}),
             Fun = fun (Q0) ->
                           TS0 = amqqueue:get_type_state(Q0),
                           TS = TS0#{nodes => [Node]},
                           amqqueue:set_type_state(Q, TS)
                   end,
             _ = rabbit_amqqueue:update(QName, Fun),
+            _ = [ra:force_delete_server(?RA_SYSTEM, {RaName, N}) || N <- OtherNodes],
             rabbit_log:warning("Disaster recovery procedure: shrinking finished");
         _ ->
             rabbit_log:warning("Disaster recovery procedure: shrinking failed, queue ~p not found at vhost ~p", [Name, VHost]),
@@ -1976,14 +1979,16 @@ force_all_queues_shrink_member_to_current_member() ->
     _ = [begin
              QName = amqqueue:get_name(Q),
              {RaName, _} = amqqueue:get_pid(Q),
+             OtherNodes = lists:delete(Node, get_nodes(Q)),
              rabbit_log:warning("Disaster recovery procedure: shrinking queue ~p", [QName]),
              ok = ra_server_proc:force_shrink_members_to_current_member({RaName, Node}),
              Fun = fun (QQ) ->
                            TS0 = amqqueue:get_type_state(QQ),
                            TS = TS0#{nodes => [Node]},
                            amqqueue:set_type_state(QQ, TS)
                    end,
-             _ = rabbit_amqqueue:update(QName, Fun)
+             _ = 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],
     rabbit_log:warning("Disaster recovery procedure: shrinking finished"),
     ok.