Always place exclusive queues on the local node

Prior to this change, exclusive queues have been subject to the queue
location process, just like other queues. Therefore, if
queue_master_locator was not client-local and x-queue-master-locator was
not set to client-local, an exclusive queue was likely to be located on
a different node than the connection it is exclusive to.  This is
suboptimal and may lead to inconsistencies when the queue's node goes
down while the connection's node is still up.

Author: mkuratczyk

deps/rabbit/src/rabbit_classic_queue.erl

@@ -55,11 +55,13 @@ is_enabled() -> true.
 declare(Q, Node) when ?amqqueue_is_classic(Q) ->
     QName = amqqueue:get_name(Q),
     VHost = amqqueue:get_vhost(Q),
-    Node1 = case Node of
-                {ignore_location, Node0} ->
+    Node1 = case {Node, rabbit_amqqueue:is_exclusive(Q)} of
+                {{ignore_location, Node0}, _} ->
                     Node0;
+                {_, true} ->
+                    Node;
                 _ ->
-                    case rabbit_queue_master_location_misc:get_location(Q)  of
+                    case rabbit_queue_master_location_misc:get_location(Q) of
                         {ok, Node0}  -> Node0;
                         _   -> Node
                     end

deps/rabbit/test/simple_ha_SUITE.erl

@@ -30,8 +30,7 @@ groups() ->
       {cluster_size_2, [], [
           rapid_redeclare,
           declare_synchrony,
-          clean_up_exclusive_queues,
-          clean_up_and_redeclare_exclusive_queues_on_other_nodes
+          clean_up_exclusive_queues
         ]},
       {cluster_size_3, [], [
           consume_survives_stop,
@@ -150,43 +149,6 @@ clean_up_exclusive_queues(Config) ->
     [[],[]] = rabbit_ct_broker_helpers:rpc_all(Config, rabbit_amqqueue, list, []),
     ok.
 
-clean_up_and_redeclare_exclusive_queues_on_other_nodes(Config) ->
-    QueueCount = 10,
-    QueueNames = lists:map(fun(N) ->
-        NBin = erlang:integer_to_binary(N),
-        <<"exclusive-q-", NBin/binary>>
-        end, lists:seq(1, QueueCount)),
-    [A, B] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
-    Conn = rabbit_ct_client_helpers:open_unmanaged_connection(Config, A),
-    {ok, Ch} = amqp_connection:open_channel(Conn),
-
-    LocationMinMasters = [
-        {<<"x-queue-master-locator">>, longstr, <<"min-masters">>}
-    ],
-    lists:foreach(fun(QueueName) ->
-            declare_exclusive(Ch, QueueName, LocationMinMasters),
-            subscribe(Ch, QueueName)
-    end, QueueNames),
-
-    ok = rabbit_ct_broker_helpers:kill_node(Config, B),
-
-    Cancels = receive_cancels([]),
-    ?assert(length(Cancels) > 0),
-
-    RemaniningQueues = rabbit_ct_broker_helpers:rpc(Config, A, rabbit_amqqueue, list, []),
-
-    ?assertEqual(length(RemaniningQueues), QueueCount - length(Cancels)),
-
-    lists:foreach(fun(QueueName) ->
-            declare_exclusive(Ch, QueueName, LocationMinMasters),
-            true = rabbit_ct_client_helpers:publish(Ch, QueueName, 1),
-            subscribe(Ch, QueueName)
-    end, QueueNames),
-    Messages = receive_messages([]),
-    ?assertEqual(10, length(Messages)),
-    ok = rabbit_ct_client_helpers:close_connection(Conn).
-
-
 consume_survives_stop(Cf)     -> consume_survives(Cf, fun stop/2,    true).
 consume_survives_sigkill(Cf)  -> consume_survives(Cf, fun sigkill/2, true).
 consume_survives_policy(Cf)   -> consume_survives(Cf, fun policy/2,  true).