Mark transient non-exclusive queues as deprecated

[Why]
Transient queues are queues that are removed upon node restart. An
application developer can't rely on such an arbitrary event to reason
about a queue's lifetime.

The only exception are exclusive transient queues which have a lifetime
linked to that of a client connection.

[How]
Non-exclusive transient queues are marked as deprecated in the code
using the Deprecated features subsystem (based on feature flags). See
pull request #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.transient_nonexcl_queues = false

Non-exclusive transient queues can be turned off anytime, there are no
conditions to do that.

Once non-exclusive transient queues are turned off, declaring a new
queue with those arguments will be rejected with a protocol error.

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.

Author: dumbbell

deps/rabbit/src/rabbit_amqqueue.erl

@@ -107,6 +107,12 @@
 
 %%----------------------------------------------------------------------------
 
+-rabbit_deprecated_feature(
+   {transient_nonexcl_queues,
+    #{deprecation_phase => permitted_by_default,
+      doc_url => "https://blog.rabbitmq.com/posts/2021/08/4.0-deprecation-announcements/#removal-of-transient-non-exclusive-queues"
+     }}).
+
 -define(CONSUMER_INFO_KEYS,
         [queue_name, channel_pid, consumer_tag, ack_required, prefetch_count,
          active, activity_status, arguments]).
@@ -223,7 +229,14 @@ declare(QueueName = #resource{virtual_host = VHost}, Durable, AutoDelete, Args,
                              VHost,
                              #{user => ActingUser},
                              Type),
-            rabbit_queue_type:declare(Q, Node);
+            case is_queue_args_combination_permitted(Q) of
+                true ->
+                    rabbit_queue_type:declare(Q, Node);
+                false ->
+                    Warning = rabbit_deprecated_features:get_warning(
+                                transient_nonexcl_queues),
+                    {protocol_error, internal_error, "~ts", [Warning]}
+            end;
         false ->
             {protocol_error, internal_error,
              "Cannot declare a queue '~ts' of type '~ts' on node '~ts': "
@@ -721,8 +734,11 @@ augment_declare_args(VHost, Durable, Exclusive, AutoDelete, Args0) ->
           when is_binary(DefaultQueueType) andalso
                not HasQTypeArg ->
             Type = rabbit_queue_type:discover(DefaultQueueType),
-            case rabbit_queue_type:is_compatible(Type, Durable,
-                                                 Exclusive, AutoDelete) of
+            IsPermitted = is_queue_args_combination_permitted(
+                            Durable, Exclusive),
+            IsCompatible = rabbit_queue_type:is_compatible(
+                             Type, Durable, Exclusive, AutoDelete),
+            case IsPermitted andalso IsCompatible of
                 true ->
                     %% patch up declare arguments with x-queue-type if there
                     %% is a vhost default set the queue is druable and not exclusive
@@ -2074,3 +2090,16 @@ get_bcc_queue(Q, BCCName) ->
     #resource{virtual_host = VHost} = amqqueue:get_name(Q),
     BCCQueueName = rabbit_misc:r(VHost, queue, BCCName),
     rabbit_amqqueue:lookup(BCCQueueName).
+
+is_queue_args_combination_permitted(Q) ->
+    Durable = amqqueue:is_durable(Q),
+    Exclusive = is_exclusive(Q),
+    is_queue_args_combination_permitted(Durable, Exclusive).
+
+is_queue_args_combination_permitted(Durable, Exclusive) ->
+    case not Durable andalso not Exclusive of
+        false ->
+            true;
+        true ->
+            rabbit_deprecated_features:is_permitted(transient_nonexcl_queues)
+    end.

deps/rabbit/test/rabbitmq_4_0_deprecations_SUITE.erl

@@ -29,7 +29,10 @@
          join_when_ram_node_type_is_not_permitted_from_conf/1,
 
          set_policy_when_cmq_is_permitted_by_default/1,
-         set_policy_when_cmq_is_not_permitted_from_conf/1
+         set_policy_when_cmq_is_not_permitted_from_conf/1,
+
+         when_transient_nonexcl_is_permitted_by_default/1,
+         when_transient_nonexcl_is_not_permitted_from_conf/1
         ]).
 
 suite() ->
@@ -39,7 +42,8 @@ all() ->
     [
      {group, global_qos},
      {group, ram_node_type},
-     {group, classic_queue_mirroring}
+     {group, classic_queue_mirroring},
+     {group, transient_nonexcl_queues}
     ].
 
 groups() ->
@@ -52,7 +56,10 @@ groups() ->
        join_when_ram_node_type_is_not_permitted_from_conf]},
      {classic_queue_mirroring, [],
       [set_policy_when_cmq_is_permitted_by_default,
-       set_policy_when_cmq_is_not_permitted_from_conf]}
+       set_policy_when_cmq_is_not_permitted_from_conf]},
+     {transient_nonexcl_queues, [],
+      [when_transient_nonexcl_is_permitted_by_default,
+       when_transient_nonexcl_is_not_permitted_from_conf]}
     ].
 
 %% -------------------------------------------------------------------
@@ -76,6 +83,8 @@ init_per_group(ram_node_type, Config) ->
                                           {rmq_nodes_clustered, false}]);
 init_per_group(classic_queue_mirroring, Config) ->
     rabbit_ct_helpers:set_config(Config, {rmq_nodes_count, 1});
+init_per_group(transient_nonexcl_queues, Config) ->
+    rabbit_ct_helpers:set_config(Config, {rmq_nodes_count, 1});
 init_per_group(_Group, Config) ->
     Config.
 
@@ -104,6 +113,14 @@ init_per_testcase(
                  [{permit_deprecated_features,
                    #{classic_queue_mirroring => false}}]}),
     init_per_testcase1(Testcase, Config1);
+init_per_testcase(
+    when_transient_nonexcl_is_not_permitted_from_conf = Testcase, Config) ->
+    Config1 = rabbit_ct_helpers:merge_app_env(
+                Config,
+                {rabbit,
+                 [{permit_deprecated_features,
+                   #{transient_nonexcl_queues => false}}]}),
+    init_per_testcase1(Testcase, Config1);
 init_per_testcase(Testcase, Config) ->
     init_per_testcase1(Testcase, Config).
 
@@ -293,3 +310,39 @@ set_policy_when_cmq_is_not_permitted_from_conf(Config) ->
          "Validation failed\n\nClassic mirrored queues are deprecated." ++ _}},
        rabbit_ct_broker_helpers:set_ha_policy(
          Config, 0, <<".*">>, <<"all">>)).
+
+%% -------------------------------------------------------------------
+%% Transient non-exclusive queues.
+%% -------------------------------------------------------------------
+
+when_transient_nonexcl_is_permitted_by_default(Config) ->
+    [NodeA] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+
+    Ch = rabbit_ct_client_helpers:open_channel(Config, NodeA),
+
+    QName = list_to_binary(atom_to_list(?FUNCTION_NAME)),
+    ?assertEqual(
+       {'queue.declare_ok', QName, 0, 0},
+       amqp_channel:call(
+         Ch,
+         #'queue.declare'{queue = QName,
+                          durable = false,
+                          exclusive = false})).
+
+when_transient_nonexcl_is_not_permitted_from_conf(Config) ->
+    [NodeA] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+
+    Ch = rabbit_ct_client_helpers:open_channel(Config, NodeA),
+
+    QName = list_to_binary(atom_to_list(?FUNCTION_NAME)),
+    ?assertExit(
+       {{shutdown,
+         {connection_closing,
+          {server_initiated_close, 541,
+           <<"INTERNAL_ERROR - Feature `transient_nonexcl_queues` is "
+             "deprecated.", _/binary>>}}}, _},
+       amqp_channel:call(
+         Ch,
+         #'queue.declare'{queue = QName,
+                          durable = false,
+                          exclusive = false})).