Deprecate queue-master-locator (#11565) (#11699)

* Deprecate queue-master-locator

This should not be a breaking change - all validation should still pass
* CQs can now use `queue-leader-locator`
* `queue-leader-locator` takes precedence over `queue-master-locator` if both are used
* regardless of which name is used, effectively there are only two  values: `client-local` (default) or `balanced`
* other values (`min-masters`, `random`, `least-leaders`) are mapped to `balanced`
* Management UI no longer shows `master-locator` fields when declaring a queue/policy, but such arguments can still be used manually (unless not permitted)
* exclusive queues are always declared locally, as before

(cherry picked from commit f398892)

Co-authored-by: Michal Kuratczyk <michal.kuratczyk@broadcom.com>

Author: mergify[bot]

deps/rabbit/BUILD.bazel

@@ -645,12 +645,6 @@ rabbitmq_integration_suite(
     size = "medium",
 )
 
-rabbitmq_integration_suite(
-    name = "queue_master_location_SUITE",
-    size = "large",
-    shard_count = 2,
-)
-
 rabbitmq_integration_suite(
     name = "queue_parallel_SUITE",
     size = "large",
@@ -678,6 +672,11 @@ rabbitmq_integration_suite(
     shard_count = 6,
 )
 
+rabbitmq_integration_suite(
+    name = "classic_queue_SUITE",
+    size = "medium",
+)
+
 rabbitmq_suite(
     name = "rabbit_confirms_SUITE",
     size = "small",
@@ -1020,6 +1019,14 @@ rabbitmq_suite(
     size = "small",
 )
 
+rabbitmq_suite(
+    name = "unit_queue_location_SUITE",
+    size = "small",
+    deps = [
+        "@meck//:erlang_app",
+    ]
+)
+
 rabbitmq_integration_suite(
     name = "unit_stats_and_metrics_SUITE",
     size = "medium",

deps/rabbit/Makefile

@@ -206,7 +206,6 @@ SLOW_CT_SUITES := backing_queue \
 		  priority_queue \
 		  priority_queue_recovery \
 		  publisher_confirms_parallel \
-		  queue_master_location \
 		  queue_parallel \
 		  quorum_queue \
 		  rabbit_core_metrics_gc \

deps/rabbit/app.bzl

@@ -14,7 +14,6 @@ def all_beam_files(name = "all_beam_files"):
             "src/rabbit_credential_validator.erl",
             "src/rabbit_exchange_type.erl",
             "src/rabbit_policy_merge_strategy.erl",
-            "src/rabbit_queue_master_locator.erl",
             "src/rabbit_queue_type.erl",
             "src/rabbit_tracking.erl",
         ],
@@ -194,11 +193,6 @@ def all_beam_files(name = "all_beam_files"):
             "src/rabbit_queue_decorator.erl",
             "src/rabbit_queue_index.erl",
             "src/rabbit_queue_location.erl",
-            "src/rabbit_queue_location_client_local.erl",
-            "src/rabbit_queue_location_min_masters.erl",
-            "src/rabbit_queue_location_random.erl",
-            "src/rabbit_queue_location_validator.erl",
-            "src/rabbit_queue_master_location_misc.erl",
             "src/rabbit_queue_type_util.erl",
             "src/rabbit_quorum_memory_manager.erl",
             "src/rabbit_quorum_queue.erl",
@@ -272,7 +266,6 @@ def all_test_beam_files(name = "all_test_beam_files"):
             "src/rabbit_credential_validator.erl",
             "src/rabbit_exchange_type.erl",
             "src/rabbit_policy_merge_strategy.erl",
-            "src/rabbit_queue_master_locator.erl",
             "src/rabbit_queue_type.erl",
             "src/rabbit_tracking.erl",
         ],
@@ -453,11 +446,6 @@ def all_test_beam_files(name = "all_test_beam_files"):
             "src/rabbit_queue_decorator.erl",
             "src/rabbit_queue_index.erl",
             "src/rabbit_queue_location.erl",
-            "src/rabbit_queue_location_client_local.erl",
-            "src/rabbit_queue_location_min_masters.erl",
-            "src/rabbit_queue_location_random.erl",
-            "src/rabbit_queue_location_validator.erl",
-            "src/rabbit_queue_master_location_misc.erl",
             "src/rabbit_queue_type_util.erl",
             "src/rabbit_quorum_memory_manager.erl",
             "src/rabbit_quorum_queue.erl",
@@ -732,12 +720,6 @@ def all_srcs(name = "all_srcs"):
             "src/rabbit_queue_decorator.erl",
             "src/rabbit_queue_index.erl",
             "src/rabbit_queue_location.erl",
-            "src/rabbit_queue_location_client_local.erl",
-            "src/rabbit_queue_location_min_masters.erl",
-            "src/rabbit_queue_location_random.erl",
-            "src/rabbit_queue_location_validator.erl",
-            "src/rabbit_queue_master_location_misc.erl",
-            "src/rabbit_queue_master_locator.erl",
             "src/rabbit_queue_type.erl",
             "src/rabbit_queue_type_util.erl",
             "src/rabbit_quorum_memory_manager.erl",
@@ -1239,15 +1221,6 @@ def test_suite_beam_files(name = "test_suite_beam_files"):
         erlc_opts = "//:test_erlc_opts",
         deps = ["//deps/amqp_client:erlang_app"],
     )
-    erlang_bytecode(
-        name = "queue_master_location_SUITE_beam_files",
-        testonly = True,
-        srcs = ["test/queue_master_location_SUITE.erl"],
-        outs = ["test/queue_master_location_SUITE.beam"],
-        app_name = "rabbit",
-        erlc_opts = "//:test_erlc_opts",
-        deps = ["//deps/amqp_client:erlang_app", "//deps/rabbitmq_ct_helpers:erlang_app"],
-    )
     erlang_bytecode(
         name = "queue_parallel_SUITE_beam_files",
         testonly = True,
@@ -2143,3 +2116,20 @@ def test_suite_beam_files(name = "test_suite_beam_files"):
         erlc_opts = "//:test_erlc_opts",
         deps = ["//deps/amqp_client:erlang_app"],
     )
+    erlang_bytecode(
+        name = "unit_queue_location_SUITE_beam_files",
+        testonly = True,
+        srcs = ["test/unit_queue_location_SUITE.erl"],
+        outs = ["test/unit_queue_location_SUITE.beam"],
+        app_name = "rabbit",
+        erlc_opts = "//:test_erlc_opts",
+    )
+    erlang_bytecode(
+        name = "classic_queue_SUITE_beam_files",
+        testonly = True,
+        srcs = ["test/classic_queue_SUITE.erl"],
+        outs = ["test/classic_queue_SUITE.beam"],
+        app_name = "rabbit",
+        erlc_opts = "//:test_erlc_opts",
+        deps = ["//deps/amqp_client:erlang_app", "//deps/rabbitmq_ct_helpers:erlang_app"],
+    )

deps/rabbit/priv/schema/rabbit.schema

@@ -1464,6 +1464,10 @@ end}.
 
 %% Queue master locator (classic queues)
 %%
+%% For backwards compatibility only as of 4.0.
+%% We still allow values of min-masters, random and client-local
+%% but the behaviour is only local or balanced.
+%% Use queue_leader_locator instead.
 
 {mapping, "queue_master_locator", "rabbit.queue_master_locator",
     [{datatype, string}]}.

deps/rabbit/src/rabbit_classic_queue.erl

@@ -1,5 +1,6 @@
 -module(rabbit_classic_queue).
 -behaviour(rabbit_queue_type).
+-behaviour(rabbit_policy_validator).
 
 -include("amqqueue.hrl").
 -include_lib("rabbit_common/include/rabbit.hrl").
@@ -63,14 +64,66 @@
          send_drained_credit_api_v1/4,
          send_credit_reply/7]).
 
+-export([validate_policy/1]).
+
+-rabbit_boot_step(
+   {?MODULE,
+    [{description, "Deprecated queue-master-locator support."
+      "Use queue-leader-locator instead."},
+     {mfa, {rabbit_registry, register,
+            [policy_validator, <<"queue-master-locator">>, ?MODULE]}},
+     {mfa, {rabbit_registry, register,
+            [operator_policy_validator, <<"queue-master-locator">>, ?MODULE]}},
+     {requires, rabbit_registry},
+     {enables, recovery}]}).
+
+validate_policy(Args) ->
+    %% queue-master-locator was deprecated in 4.0
+    Locator = proplists:get_value(<<"queue-master-locator">>, Args, unknown),
+    case Locator of
+        unknown ->
+            ok;
+        _ ->
+            case rabbit_queue_location:master_locator_permitted() of
+                true ->
+                    case lists:member(Locator, rabbit_queue_location:queue_leader_locators()) of
+                        true -> ok;
+                        false -> {error, "~tp is not a valid master locator", [Locator]}
+                    end;
+                false ->
+                    {error, "use of deprecated queue-master-locator argument is not permitted", []}
+                end
+    end.
+
 -spec is_enabled() -> boolean().
 is_enabled() -> true.
 
 -spec is_compatible(boolean(), boolean(), boolean()) -> boolean().
 is_compatible(_, _, _) ->
     true.
 
+validate_arguments(Args) ->
+    case lists:keymember(<<"x-queue-master-locator">>, 1, Args) of
+        false ->
+            ok;
+        true ->
+            case rabbit_queue_location:master_locator_permitted() of
+                true ->
+                    ok;
+                false ->
+                    Warning = rabbit_deprecated_features:get_warning(
+                                queue_master_locator),
+                    {protocol_error, internal_error, "~ts", [Warning]}
+            end
+    end.
+
 declare(Q, Node) when ?amqqueue_is_classic(Q) ->
+    case validate_arguments(amqqueue:get_arguments(Q)) of
+        ok -> do_declare(Q, Node);
+        Error -> Error
+    end.
+
+do_declare(Q, Node) when ?amqqueue_is_classic(Q) ->
     QName = amqqueue:get_name(Q),
     VHost = amqqueue:get_vhost(Q),
     Node1 = case {Node, rabbit_amqqueue:is_exclusive(Q)} of
@@ -79,10 +132,8 @@ declare(Q, Node) when ?amqqueue_is_classic(Q) ->
                 {_, true} ->
                     Node;
                 _ ->
-                    case rabbit_queue_master_location_misc:get_location(Q) of
-                        {ok, Node0}  -> Node0;
-                        _   -> Node
-                    end
+                    {Node0, _} = rabbit_queue_location:select_leader_and_followers(Q, 1),
+                    Node0
             end,
     case rabbit_vhost_sup_sup:get_vhost_sup(VHost, Node1) of
         {ok, _} ->
@@ -509,15 +560,18 @@ recover_durable_queues(QueuesAndRecoveryTerms) ->
 capabilities() ->
     #{unsupported_policies => [%% Stream policies
                                <<"max-age">>, <<"stream-max-segment-size-bytes">>,
-                               <<"queue-leader-locator">>, <<"initial-cluster-size">>,
+                               <<"initial-cluster-size">>,
                                %% Quorum policies
                                <<"delivery-limit">>, <<"dead-letter-strategy">>, <<"max-in-memory-length">>, <<"max-in-memory-bytes">>, <<"target-group-size">>],
       queue_arguments => [<<"x-expires">>, <<"x-message-ttl">>, <<"x-dead-letter-exchange">>,
                           <<"x-dead-letter-routing-key">>, <<"x-max-length">>,
                           <<"x-max-length-bytes">>, <<"x-max-priority">>,
                           <<"x-overflow">>, <<"x-queue-mode">>, <<"x-queue-version">>,
-                          <<"x-single-active-consumer">>, <<"x-queue-type">>,
-                          <<"x-queue-master-locator">>],
+                          <<"x-single-active-consumer">>, <<"x-queue-type">>, <<"x-queue-master-locator">>]
+                          ++ case rabbit_feature_flags:is_enabled(classic_queue_leader_locator) of
+                                 true -> [<<"x-queue-leader-locator">>];
+                                 false -> []
+                             end,
       consumer_arguments => [<<"x-priority">>, <<"x-credit">>],
       server_named => true}.
 

deps/rabbit/src/rabbit_core_ff.erl

@@ -185,3 +185,10 @@
       stability     => stable,
       depends_on    => [message_containers]
      }}).
+
+-rabbit_feature_flag(
+   {classic_queue_leader_locator,
+    #{desc          => "queue-leader-locator support in classic queues",
+      doc_url       => "https://www.rabbitmq.com/docs/clustering#replica-placement",
+      stability     => stable
+     }}).