Mark global QoS setting as deprecated

dumbbell · dumbbell · commit a7eda955d85a · 2023-07-04T16:46:39.000+02:00
[Why] Global QoS, where a single shared prefetch is used for an entire channel, is not recommended practice. Per-consumer QoS (non-global) should be set instead. [How] The global QoS setting is marked as deprecated in the code using the Deprecated features subsystem (based on feature flags). See #7390 for a description of that subsystem. To test RabbitMQ behavior as if the feature was removed, the following configuration setting can be used: deprecated_features.permit.global_qos = false Global QoS can be turned off anytime, there are no conditions to do that. Once global QoS is turned off, the prefetch setting will always be considered as non-global (i.e. per-consumer). A warning message will be logged if the default prefetch setting enables global QoS or anytime a client requests a global QoS on the channel. Note that given the marketing calendar, the deprecated feature will go directly from "permitted by default" to "removed" in RabbitMQ 4.0. It won't go through the gradual deprecation process.
diff --git a/deps/rabbit/BUILD.bazel b/deps/rabbit/BUILD.bazel
@@ -803,6 +803,11 @@ rabbitmq_integration_suite(
     ],
 )
 
+rabbitmq_integration_suite(
+    name = "rabbitmq_4_0_deprecations_SUITE",
+    size = "large",
+)
+
 rabbitmq_integration_suite(
     name = "rabbitmq_queues_cli_integration_SUITE",
     size = "medium",
diff --git a/deps/rabbit/app.bzl b/deps/rabbit/app.bzl
@@ -1963,3 +1963,12 @@ def test_suite_beam_files(name = "test_suite_beam_files"):
         erlc_opts = "//:test_erlc_opts",
         deps = ["//deps/amqp_client:erlang_app"],
     )
+    erlang_bytecode(
+        name = "rabbitmq_4_0_deprecations_SUITE_beam_files",
+        testonly = True,
+        srcs = ["test/rabbitmq_4_0_deprecations_SUITE.erl"],
+        outs = ["test/rabbitmq_4_0_deprecations_SUITE.beam"],
+        app_name = "rabbit",
+        erlc_opts = "//:test_erlc_opts",
+        deps = ["//deps/amqp_client:erlang_app"],
+    )
diff --git a/deps/rabbit/src/rabbit_channel.erl b/deps/rabbit/src/rabbit_channel.erl
@@ -236,6 +236,12 @@
 
 %%----------------------------------------------------------------------------
 
+-rabbit_deprecated_feature(
+   {global_qos,
+    #{deprecation_phase => permitted_by_default,
+      doc_url => "https://blog.rabbitmq.com/posts/2021/08/4.0-deprecation-announcements/#removal-of-global-qos"
+     }}).
+
 -spec start_link
         (channel_number(), pid(), pid(), pid(), string(), rabbit_types:protocol(),
          rabbit_types:user(), rabbit_types:vhost(), rabbit_framing:amqp_table(),
@@ -506,8 +512,9 @@ init([Channel, ReaderPid, WriterPid, ConnPid, ConnName, Protocol, User, VHost,
              true   -> flow;
              false  -> noflow
            end,
-    {ok, {Global, Prefetch}} = application:get_env(rabbit, default_consumer_prefetch),
+    {ok, {Global0, Prefetch}} = application:get_env(rabbit, default_consumer_prefetch),
     Limiter0 = rabbit_limiter:new(LimiterPid),
+    Global = Global0 andalso is_global_qos_permitted(),
     Limiter = case {Global, Prefetch} of
                   {true, 0} ->
                       rabbit_limiter:unlimit_prefetch(Limiter0);
@@ -1559,30 +1566,44 @@ handle_method(#'basic.qos'{global         = false,
                                         limiter = Limiter1}};
 
 handle_method(#'basic.qos'{global         = true,
-                           prefetch_count = 0},
-              _, State = #ch{limiter = Limiter}) ->
-    Limiter1 = rabbit_limiter:unlimit_prefetch(Limiter),
-    case rabbit_limiter:is_active(Limiter) of
-        true  -> rabbit_amqqueue:deactivate_limit_all(
-                   classic_consumer_queue_pids(State#ch.consumer_mapping), self());
-        false -> ok
-    end,
-    {reply, #'basic.qos_ok'{}, State#ch{limiter = Limiter1}};
+                           prefetch_count = 0} = Method,
+              Content,
+              State = #ch{limiter = Limiter}) ->
+    case is_global_qos_permitted() of
+        true ->
+            Limiter1 = rabbit_limiter:unlimit_prefetch(Limiter),
+            case rabbit_limiter:is_active(Limiter) of
+                true  -> rabbit_amqqueue:deactivate_limit_all(
+                           classic_consumer_queue_pids(State#ch.consumer_mapping), self());
+                false -> ok
+            end,
+            {reply, #'basic.qos_ok'{}, State#ch{limiter = Limiter1}};
+        false ->
+            Method1 = Method#'basic.qos'{global = false},
+            handle_method(Method1, Content, State)
+    end;
 
 handle_method(#'basic.qos'{global         = true,
-                           prefetch_count = PrefetchCount},
-              _, State = #ch{limiter = Limiter, unacked_message_q = UAMQ}) ->
-    %% TODO ?QUEUE:len(UAMQ) is not strictly right since that counts
-    %% unacked messages from basic.get too. Pretty obscure though.
-    Limiter1 = rabbit_limiter:limit_prefetch(Limiter,
-                                             PrefetchCount, ?QUEUE:len(UAMQ)),
-    case ((not rabbit_limiter:is_active(Limiter)) andalso
-          rabbit_limiter:is_active(Limiter1)) of
-        true  -> rabbit_amqqueue:activate_limit_all(
-                   classic_consumer_queue_pids(State#ch.consumer_mapping), self());
-        false -> ok
-    end,
-    {reply, #'basic.qos_ok'{}, State#ch{limiter = Limiter1}};
+                           prefetch_count = PrefetchCount} = Method,
+              Content,
+              State = #ch{limiter = Limiter, unacked_message_q = UAMQ}) ->
+    case is_global_qos_permitted() of
+        true ->
+            %% TODO ?QUEUE:len(UAMQ) is not strictly right since that counts
+            %% unacked messages from basic.get too. Pretty obscure though.
+            Limiter1 = rabbit_limiter:limit_prefetch(Limiter,
+                                                     PrefetchCount, ?QUEUE:len(UAMQ)),
+            case ((not rabbit_limiter:is_active(Limiter)) andalso
+                  rabbit_limiter:is_active(Limiter1)) of
+                true  -> rabbit_amqqueue:activate_limit_all(
+                           classic_consumer_queue_pids(State#ch.consumer_mapping), self());
+                false -> ok
+            end,
+            {reply, #'basic.qos_ok'{}, State#ch{limiter = Limiter1}};
+        false ->
+            Method1 = Method#'basic.qos'{global = false},
+            handle_method(Method1, Content, State)
+    end;
 
 handle_method(#'basic.recover_async'{requeue = true},
               _, State = #ch{unacked_message_q = UAMQ,
@@ -2943,3 +2964,6 @@ maybe_decrease_global_publishers(#ch{publishing_mode = false}) ->
     ok;
 maybe_decrease_global_publishers(#ch{publishing_mode = true}) ->
     rabbit_global_counters:publisher_deleted(amqp091).
+
+is_global_qos_permitted() ->
+    rabbit_deprecated_features:is_permitted(global_qos).
diff --git a/deps/rabbit/test/rabbitmq_4_0_deprecations_SUITE.erl b/deps/rabbit/test/rabbitmq_4_0_deprecations_SUITE.erl
@@ -0,0 +1,141 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2023 VMware, Inc. or its affiliates.  All rights reserved.
+%%
+
+-module(rabbitmq_4_0_deprecations_SUITE).
+
+-include_lib("eunit/include/eunit.hrl").
+-include_lib("common_test/include/ct.hrl").
+
+-include_lib("amqp_client/include/amqp_client.hrl").
+
+-export([suite/0,
+         all/0,
+         groups/0,
+         init_per_suite/1,
+         end_per_suite/1,
+         init_per_group/2,
+         end_per_group/2,
+         init_per_testcase/2,
+         end_per_testcase/2,
+
+         when_global_qos_is_permitted_by_default/1,
+         when_global_qos_is_not_permitted_from_conf/1
+        ]).
+
+suite() ->
+    [{timetrap, {minutes, 5}}].
+
+all() ->
+    [
+     {group, global_qos}
+    ].
+
+groups() ->
+    [
+     {global_qos, [],
+      [when_global_qos_is_permitted_by_default,
+       when_global_qos_is_not_permitted_from_conf]}
+    ].
+
+%% -------------------------------------------------------------------
+%% Testsuite setup/teardown.
+%% -------------------------------------------------------------------
+
+init_per_suite(Config) ->
+    rabbit_ct_helpers:log_environment(),
+    logger:set_primary_config(level, debug),
+    rabbit_ct_helpers:run_setup_steps(
+      Config,
+      [fun rabbit_ct_helpers:redirect_logger_to_ct_logs/1]).
+
+end_per_suite(Config) ->
+    Config.
+
+init_per_group(global_qos, Config) ->
+    rabbit_ct_helpers:set_config(Config, {rmq_nodes_count, 1});
+init_per_group(_Group, Config) ->
+    Config.
+
+end_per_group(_Group, Config) ->
+    Config.
+
+init_per_testcase(
+  when_global_qos_is_not_permitted_from_conf = Testcase, Config) ->
+    Config1 = rabbit_ct_helpers:merge_app_env(
+                Config,
+                {rabbit,
+                 [{permit_deprecated_features, #{global_qos => false}}]}),
+    init_per_testcase1(Testcase, Config1);
+init_per_testcase(Testcase, Config) ->
+    init_per_testcase1(Testcase, Config).
+
+init_per_testcase1(Testcase, Config) ->
+    rabbit_ct_helpers:testcase_started(Config, Testcase),
+    ClusterSize = ?config(rmq_nodes_count, Config),
+    TestNumber = rabbit_ct_helpers:testcase_number(Config, ?MODULE, Testcase),
+    Config1 = rabbit_ct_helpers:set_config(Config, [
+        {rmq_nodename_suffix, Testcase},
+        {tcp_ports_base, {skip_n_nodes, TestNumber * ClusterSize}},
+        {keep_pid_file_on_exit, true}
+      ]),
+    rabbit_ct_helpers:run_steps(Config1,
+      rabbit_ct_broker_helpers:setup_steps() ++
+      rabbit_ct_client_helpers:setup_steps()).
+
+end_per_testcase(Testcase, Config) ->
+    Config1 = rabbit_ct_helpers:run_steps(Config,
+      rabbit_ct_client_helpers:teardown_steps() ++
+      rabbit_ct_broker_helpers:teardown_steps()),
+    rabbit_ct_helpers:testcase_finished(Config1, Testcase).
+
+%% -------------------------------------------------------------------
+%% Global QoS.
+%% -------------------------------------------------------------------
+
+when_global_qos_is_permitted_by_default(Config) ->
+    [NodeA] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+
+    ExistingServerChs = list_server_channels(Config, NodeA),
+    ClientCh = rabbit_ct_client_helpers:open_channel(Config, NodeA),
+    [ServerCh] = list_server_channels(Config, NodeA) -- ExistingServerChs,
+
+    ?assertNot(is_prefetch_limited(ServerCh)),
+
+    %% It's possible to request global QoS and it is accepted by the server.
+    ?assertMatch(
+       #'basic.qos_ok'{},
+       amqp_channel:call(
+         ClientCh,
+         #'basic.qos'{global = true, prefetch_count = 10})),
+    ?assert(is_prefetch_limited(ServerCh)).
+
+when_global_qos_is_not_permitted_from_conf(Config) ->
+    [NodeA] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+
+    ExistingServerChs = list_server_channels(Config, NodeA),
+    ClientCh = rabbit_ct_client_helpers:open_channel(Config, NodeA),
+    [ServerCh] = list_server_channels(Config, NodeA) -- ExistingServerChs,
+
+    ?assertNot(is_prefetch_limited(ServerCh)),
+
+    %% It's possible to request global QoS but it is ignored by the server.
+    ?assertMatch(
+       #'basic.qos_ok'{},
+       amqp_channel:call(
+         ClientCh,
+         #'basic.qos'{global = true, prefetch_count = 10})),
+    ?assertNot(is_prefetch_limited(ServerCh)).
+
+list_server_channels(Config, Node) ->
+    rabbit_ct_broker_helpers:rpc(Config, Node, rabbit_channel, list, []).
+
+is_prefetch_limited(ServerCh) ->
+    GenServer2State = sys:get_state(ServerCh),
+    ChState = element(4, GenServer2State),
+    ct:pal("Server channel (~p) state: ~p", [ServerCh, ChState]),
+    LimiterState = element(3, ChState),
+    element(3, LimiterState).
