QQ: when invoking drain only shut down small batches at a time

Then wait for elections to complete before shutting further
members down.

This should help avoid election storms when enabling maintenance
mode.

Author: kjnilsson

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -1582,12 +1582,12 @@ transfer_leadership(Q, Destination) ->
     {RaName, _} = Pid = amqqueue:get_pid(Q),
     case ra:transfer_leadership(Pid, {RaName, Destination}) of
         ok ->
-          case ra:members(Pid) of
-            {_, _, {_, NewNode}} ->
-              {migrated, NewNode};
-            {timeout, _} ->
-              {not_migrated, ra_members_timeout}
-          end;
+            case ra:members(Pid) of
+                {_, _, {_, NewNode}} ->
+                    {migrated, NewNode};
+                {timeout, _} ->
+                    {not_migrated, ra_members_timeout}
+            end;
         already_leader ->
             {not_migrated, already_leader};
         {error, Reason} ->
@@ -2299,27 +2299,51 @@ drain(TransferCandidates) ->
 transfer_leadership([]) ->
     ?LOG_WARNING("Skipping leadership transfer of quorum queues: no candidate "
                        "(online, not under maintenance) nodes to transfer to!");
-transfer_leadership(_TransferCandidates) ->
+transfer_leadership(_CandidateNodes) ->
     %% we only transfer leadership for QQs that have local leaders
-    Queues = rabbit_amqqueue:list_local_leaders(),
+    LocalLeaderQueues = rabbit_amqqueue:list_local_leaders(),
+    QueuesChunked = ra_lib:lists_chunk(16, LocalLeaderQueues),
     ?LOG_INFO("Will transfer leadership of ~b quorum queues with current leader on this node",
-                    [length(Queues)]),
-    _ = [begin
-        Name = amqqueue:get_name(Q),
-        ?LOG_DEBUG("Will trigger a leader election for local quorum queue ~ts",
-                         [rabbit_misc:rs(Name)]),
-        %% we trigger an election and exclude this node from the list of candidates
-        %% by simply shutting its local QQ replica (Ra server)
-        RaLeader = amqqueue:get_pid(Q),
-        ?LOG_DEBUG("Will stop Ra server ~tp", [RaLeader]),
-        case rabbit_quorum_queue:stop_server(RaLeader) of
-            ok     ->
-                ?LOG_DEBUG("Successfully stopped Ra server ~tp", [RaLeader]);
-            {error, nodedown} ->
-                ?LOG_ERROR("Failed to stop Ra server ~tp: target node was reported as down")
-        end
-     end || Q <- Queues],
-    ?LOG_INFO("Leadership transfer for quorum queues hosted on this node has been initiated").
+              [length(LocalLeaderQueues)]),
+    [begin
+         [begin
+              % ?LOG_DEBUG("Will trigger a leader election for local quorum queue ~ts",
+              %            [rabbit_misc:rs(Name)]),
+              %% we trigger an election and exclude this node from the list of candidates
+              %% by simply shutting its local QQ replica (Ra server)
+              RaLeader = amqqueue:get_pid(Q),
+              ?LOG_DEBUG("Will stop Ra leader ~tp", [RaLeader]),
+              case rabbit_quorum_queue:stop_server(RaLeader) of
+                  ok     ->
+                      ?LOG_DEBUG("Successfully stopped Ra server ~tp", [RaLeader]);
+                  {error, nodedown} ->
+                      ?LOG_ERROR("Failed to stop Ra server ~tp: target node was reported as down")
+              end,
+              ok
+          end || Q <- Queues],
+         %% wait for leader elections before processing next chunk of queues
+         [begin
+              {RaName, LeaderNode} = amqqueue:get_pid(Q),
+              MemberNodes = lists:delete(LeaderNode, amqqueue:get_nodes(Q)),
+              %% we don't do any explicit error handling here
+              _ = lists:any(fun (N) ->
+                                    case ra:members({RaName, N}, 30000) of
+                                        {ok, _, _} ->
+                                            true;
+                                        Err ->
+                                            Name = amqqueue:get_name(Q),
+                                            ?LOG_DEBUG("Failed to wait for leader election for queue ~ts on ~tp Err ~ts",
+                                                       [Name, N, Err]),
+                                            false
+                                    end
+                            end, MemberNodes),
+              ok
+
+          end || Q <- Queues],
+         ok
+     end || Queues <- QueuesChunked],
+    ?LOG_INFO("Leadership transfer for quorum queues hosted on this node has been initiated"),
+    ok.
 
 %% TODO: I just copied it over, it looks like was always called inside maintenance so...
 -spec stop_local_quorum_queue_followers() -> ok.