Fix migrate leadership timeout

The code of `migrate_leadership_to_existing_replica` previously assumed
that it can control the target node of a failover of mirrored queues.

While its sibling function `transfer_leadership` can do that (as it drops all
mirrors besides the one on the target node)
`migrate_leadership_to_existing_replica` tries to be less intrusive and
only tries to migrate the queue away for the current primary.

However it then used `wait_for_new_master` to wait for the queue to actually
transfer the primary to the specified target node. As the specified target
node might not ever become the primary of the queue (as this decission is
only made between the remaining mirrors) this can cause the migrate operation
to run into a timeout (10 sec per default).

As `migrate_leadership_to_existing_replica` is only used during
`transfer_leadership_of_classic_mirrored_queues` its only goal is to get the
primary away from the current node. Therefor we can just wait for the
queue to become active on some other node instead of expecting a specific
node to become the primary.

(cherry picked from commit fa0a8de)

Author: felixhuettner

deps/rabbit/src/rabbit_maintenance.erl

@@ -270,15 +270,15 @@ transfer_leadership_of_classic_mirrored_queues(TransferCandidates) ->
                           [rabbit_misc:rs(Name), readable_candidate_list(ExistingReplicaNodes)]),
          case random_primary_replica_transfer_candidate_node(TransferCandidates, ExistingReplicaNodes) of
              {ok, Pick} ->
-                 rabbit_log:debug("Will transfer leadership of local ~s to node ~s",
+                 rabbit_log:debug("Will transfer leadership of local ~s. Planned target node: ~s",
                           [rabbit_misc:rs(Name), Pick]),
                  case rabbit_mirror_queue_misc:migrate_leadership_to_existing_replica(Q, Pick) of
-                     {migrated, _} ->
+                     {migrated, NewPrimary} ->
                          rabbit_log:debug("Successfully transferred leadership of queue ~s to node ~s",
-                                          [rabbit_misc:rs(Name), Pick]);
+                                          [rabbit_misc:rs(Name), NewPrimary]);
                      Other ->
-                         rabbit_log:warning("Could not transfer leadership of queue ~s to node ~s: ~p",
-                                            [rabbit_misc:rs(Name), Pick, Other])
+                         rabbit_log:warning("Could not transfer leadership of queue ~s: ~p",
+                                            [rabbit_misc:rs(Name), Other])
                  end;
              undefined ->
                  rabbit_log:warning("Could not transfer leadership of queue ~s: no suitable candidates?",

deps/rabbit/src/rabbit_mirror_queue_misc.erl

@@ -602,6 +602,8 @@ transfer_leadership(Q, Destination) ->
 %% Moves the primary replica (leader) of a classic mirrored queue to another node
 %% which already hosts a replica of this queue. In this case we can stop
 %% fewer replicas and reduce the load the operation has on the cluster.
+%% Note that there is no guarantee that the queue will actually end up on the
+%% destination node. The actual destination node is returned.
 migrate_leadership_to_existing_replica(Q, Destination) ->
     QName = amqqueue:get_name(Q),
     {PreTransferPrimaryNode, PreTransferMirrorNodes, _PreTransferInSyncMirrorNodes} = actual_queue_nodes(Q),
@@ -616,7 +618,7 @@ migrate_leadership_to_existing_replica(Q, Destination) ->
     NodesToDropMirrorsOn = [PreTransferPrimaryNode],
     drop_mirrors(QName, NodesToDropMirrorsOn),
 
-    case wait_for_new_master(QName, Destination) of
+    case wait_for_different_master(QName, PreTransferPrimaryNode) of
         not_migrated ->
             {not_migrated, undefined};
         {{not_migrated, Destination} = Result, _Q1} ->
@@ -655,6 +657,36 @@ wait_for_new_master(QName, Destination, N) ->
             end
     end.
 
+-spec wait_for_different_master(rabbit_amqqueue:name(), atom()) -> {{migrated, node()}, amqqueue:amqqueue()} | {{not_migrated, node()}, amqqueue:amqqueue()} | not_migrated.
+wait_for_different_master(QName, Source) ->
+    wait_for_different_master(QName, Source, 100).
+
+wait_for_different_master(QName, _, 0) ->
+    case rabbit_amqqueue:lookup(QName) of
+        {error, not_found} -> not_migrated;
+        {ok, Q}            -> {{not_migrated, undefined}, Q}
+    end;
+wait_for_different_master(QName, Source, N) ->
+    case rabbit_amqqueue:lookup(QName) of
+        {error, not_found} ->
+            not_migrated;
+        {ok, Q} ->
+            case amqqueue:get_pid(Q) of
+                none ->
+                    timer:sleep(100),
+                    wait_for_different_master(QName, Source, N - 1);
+                Pid ->
+                    case node(Pid) of
+                        Source ->
+                            timer:sleep(100),
+                            wait_for_different_master(QName, Source, N - 1);
+                        Destination ->
+                            {{migrated, Destination}, Q}
+                    end
+            end
+    end.
+
+
 %% The arrival of a newly synced mirror may cause the master to die if
 %% the policy does not want the master but it has been kept alive
 %% because there were no synced mirrors.