Merge pull request #3341 from rabbitmq/local-exclusive-queues

Always place exclusive queues on the local node

Author: michaelklishin

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).