Merge pull request #4486 from rabbitmq/mergify/bp/v3.9.x/pr-4481

Fix migrate leadership timeout (backport #4468) (backport #4481)

Author: michaelklishin

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

@@ -582,6 +582,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),
@@ -596,7 +598,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} ->
@@ -635,6 +637,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.