Remove feature flag compat code for rabbitmq_4.2.0

ansd · ansd · commit 2a9ba1959b76 · 2026-03-19T09:54:18.000+01:00
diff --git a/deps/rabbit/src/rabbit_amqp_session.erl b/deps/rabbit/src/rabbit_amqp_session.erl
@@ -2822,14 +2822,12 @@ ensure_source(#'v1_0.source'{
                 } = Source0,
               SndSettled, LinkName, Vhost, User, ContainerId,
               ConnPid, PermCache0, TopicPermCache) ->
-    FFEnabled = rabbit_volatile_queue:ff_enabled(),
     case maps:from_keys(Caps, true) of
         #{{symbol, ?CAP_VOLATILE_QUEUE} := true}
           when (Durable =:= undefined orelse Durable =:= ?V_1_0_TERMINUS_DURABILITY_NONE) andalso
                ExpiryPolicy =:= ?V_1_0_TERMINUS_EXPIRY_POLICY_LINK_DETACH andalso
                (Timeout =:= undefined orelse Timeout =:= {uint, 0}) andalso
-               SndSettled andalso
-               FFEnabled ->
+               SndSettled ->
             %% create volatile queue
             QNameBin = rabbit_volatile_queue:new_name(),
             Source = #'v1_0.source'{
diff --git a/deps/rabbit/src/rabbit_channel.erl b/deps/rabbit/src/rabbit_channel.erl
@@ -63,7 +63,7 @@
          format_state/1, format_message_queue/2]).
 
 %% Internal
--export([list_local/0, emit_info_local/3, deliver_reply_local/3]).
+-export([list_local/0, emit_info_local/3]).
 -export([get_vhost/1, get_user/1]).
 %% For testing
 -export([build_topic_variable_map/3]).
@@ -298,14 +298,6 @@ shutdown(Pid) ->
 send_command(Pid, Msg) ->
     gen_server2:cast(Pid,  {command, Msg}).
 
-%% Delete this function when feature flag rabbitmq_4.2.0 becomes required.
--spec deliver_reply_local(pid(), binary(), mc:state()) -> ok.
-deliver_reply_local(Pid, Key, Message) ->
-    case pg_local:in_group(rabbit_channels, Pid) of
-        true  -> gen_server2:cast(Pid, {deliver_reply, Key, Message});
-        false -> ok
-    end.
-
 -spec list() -> [pid()].
 
 list() ->
@@ -561,20 +553,6 @@ handle_call({has_state, #resource{virtual_host = Vhost,
     reply(true, State);
 handle_call({has_state, _QName, _QType}, _From, State) ->
     reply(false, State);
-%% Delete below clause when feature flag rabbitmq_4.2.0 becomes required.
-handle_call({declare_fast_reply_to, Key}, _From, State = #ch{direct_reply = Reply}) ->
-    Result = case Reply of
-                 none ->
-                     not_found;
-                 #direct_reply{queue = QNameBin} ->
-                     case rabbit_volatile_queue:key_from_name(QNameBin) of
-                         {ok, Key} ->
-                             exists;
-                         _ ->
-                             not_found
-                     end
-             end,
-    reply(Result, State);
 
 handle_call(refresh_config, _From,
             State = #ch{cfg = #conf{virtual_host = VHost} = Cfg}) ->
@@ -647,30 +625,6 @@ handle_cast({command, Msg}, State) ->
     ok = send(Msg, State),
     noreply(State);
 
-%% Delete below clause when feature flag rabbitmq_4.2.0 becomes required.
-handle_cast({deliver_reply, Key, Mc},
-            #ch{cfg = #conf{state = ChanState,
-                            writer_pid = WriterPid,
-                            msg_interceptor_ctx = MsgIcptCtx},
-                next_tag = DeliveryTag,
-                direct_reply = #direct_reply{consumer_tag = Ctag,
-                                             queue = QNameBin}} = State)
-  when ChanState =/= closing ->
-    case rabbit_volatile_queue:key_from_name(QNameBin) of
-        {ok, Key} ->
-            ExchName = mc:exchange(Mc),
-            [RoutingKey | _] = mc:routing_keys(Mc),
-            Deliver = #'basic.deliver'{consumer_tag = Ctag,
-                                       delivery_tag = DeliveryTag,
-                                       redelivered = false,
-                                       exchange = ExchName,
-                                       routing_key = RoutingKey},
-            Content = outgoing_content(Mc, MsgIcptCtx),
-            ok = rabbit_writer:send_command(WriterPid, Deliver, Content);
-        _ ->
-            ok
-    end,
-    noreply(State);
 handle_cast({deliver_reply, _, _}, State) ->
     noreply(State);
 
diff --git a/deps/rabbit/src/rabbit_volatile_queue.erl b/deps/rabbit/src/rabbit_volatile_queue.erl
@@ -20,10 +20,8 @@
          new_target/1,
          new_name/0,
          is/1,
-         key_from_name/1,
          pid_from_name/2,
          exists/1,
-         ff_enabled/0,
          local_cast/2,
          local_call/2]).
 
@@ -159,25 +157,10 @@ exists(#resource{kind = queue,
                  name = QNameBin} = QName) ->
     case pid_from_name(QNameBin) of
         {ok, Pid} when is_pid(Pid) ->
-            case ff_enabled() of
-                true ->
-                    Request = {has_state, QName, ?MODULE},
-                    MFA = {?MODULE, local_call, [Request]},
-                    try delegate:invoke(Pid, MFA)
-                    catch _:_ -> false
-                    end;
-                false ->
-                    case key_from_name(QNameBin) of
-                        {ok, Key} ->
-                            Msg = {declare_fast_reply_to, Key},
-                            try gen_server:call(Pid, Msg, infinity) of
-                                exists -> true;
-                                _ -> false
-                            catch exit:_ -> false
-                            end;
-                        error ->
-                            false
-                    end
+            Request = {has_state, QName, ?MODULE},
+            MFA = {?MODULE, local_call, [Request]},
+            try delegate:invoke(Pid, MFA)
+            catch _:_ -> false
             end;
         _ ->
             false
@@ -201,22 +184,11 @@ deliver(Qs, Msg, #{}) ->
     {[], []}.
 
 deliver0(Q, Msg) ->
-    QName = amqqueue:get_name(Q),
     QPid = amqqueue:get_pid(Q),
-    case ff_enabled() of
-        true ->
-            Request = {queue_event, QName, {deliver, Msg}},
-            MFA = {?MODULE, local_cast, [Request]},
-            delegate:invoke_no_result(QPid, MFA);
-        false ->
-            case key_from_name(QName#resource.name) of
-                {ok, Key} ->
-                    MFA = {rabbit_channel, deliver_reply_local, [Key, Msg]},
-                    delegate:invoke_no_result(QPid, MFA);
-                error ->
-                    ok
-            end
-    end.
+    QName = amqqueue:get_name(Q),
+    Request = {queue_event, QName, {deliver, Msg}},
+    MFA = {?MODULE, local_cast, [Request]},
+    delegate:invoke_no_result(QPid, MFA).
 
 -spec local_cast(pid(), term()) -> ok.
 local_cast(Pid, Request) ->
@@ -290,9 +262,6 @@ cancel(_, _, #?STATE{} = State) ->
 is_enabled() ->
     true.
 
-ff_enabled() ->
-    rabbit_feature_flags:is_enabled('rabbitmq_4.2.0').
-
 is_compatible(_, _, _) ->
     true.
 
@@ -422,18 +391,5 @@ pid_from_name(<<?PREFIX, Bin/binary>>, CandidateNodes) ->
 pid_from_name(_, _) ->
     error.
 
-%% Returns the base 64 encoded key.
--spec key_from_name(rabbit_misc:resource_name()) ->
-    {ok, binary()} | error.
-key_from_name(<<?PREFIX, Suffix/binary>>) ->
-    case binary:split(Suffix, <<".">>) of
-        [_Pid, Key] ->
-            {ok, Key};
-        _ ->
-            error
-    end;
-key_from_name(_) ->
-    error.
-
 nodes_with_hashes() ->
     #{erlang:phash2(Node) => Node || Node <- rabbit_nodes:list_members()}.
diff --git a/deps/rabbit/test/amqp_auth_SUITE.erl b/deps/rabbit/test/amqp_auth_SUITE.erl
@@ -482,8 +482,6 @@ attach_source_queue_dynamic_exclusive(Config) ->
     ok = close_connection_sync(Connection).
 
 attach_source_queue_dynamic_volatile(Config) ->
-    ok = rabbit_ct_broker_helpers:enable_feature_flag(Config, 'rabbitmq_4.2.0'),
-
     OpnConf = connection_config(Config),
     {ok, Connection} = amqp10_client:open_connection(OpnConf),
     {ok, Session} = amqp10_client:begin_session_sync(Connection),
diff --git a/deps/rabbit/test/clustering_recovery_SUITE.erl b/deps/rabbit/test/clustering_recovery_SUITE.erl
@@ -139,39 +139,10 @@ init_per_testcase(Testcase, Config) ->
         {tcp_ports_base, {skip_n_nodes, TestNumber * ClusterSize}},
         {keep_pid_file_on_exit, true}
       ]),
-    Config2 = rabbit_ct_helpers:run_steps(
-                Config1,
-                rabbit_ct_broker_helpers:setup_steps() ++
-                rabbit_ct_client_helpers:setup_steps()),
-    case Config2 of
-        _ when is_list(Config2) andalso
-               (Testcase =:= autodelete_transient_queue_after_partition_recovery_1 orelse
-                Testcase =:= autodelete_durable_queue_after_partition_recovery_1 orelse
-                Testcase =:= autodelete_transient_queue_after_partition_recovery_2 orelse
-                Testcase =:= autodelete_durable_queue_after_partition_recovery_2 orelse
-                Testcase =:= exclusive_transient_queue_after_partition_recovery_1 orelse
-                Testcase =:= exclusive_durable_queue_after_partition_recovery_1 orelse
-                Testcase =:= exclusive_transient_queue_after_partition_recovery_2 orelse
-                Testcase =:= exclusive_durable_queue_after_partition_recovery_2) ->
-            NewEnough = ok =:= rabbit_ct_broker_helpers:enable_feature_flag(
-                                 Config2, 'rabbitmq_4.2.0'),
-            case NewEnough of
-                true ->
-                    Config2;
-                false ->
-                    _ = rabbit_ct_helpers:run_steps(
-                          Config2,
-                          rabbit_ct_client_helpers:teardown_steps() ++
-                          rabbit_ct_broker_helpers:teardown_steps()),
-                    rabbit_ct_helpers:testcase_finished(Config2, Testcase),
-                    {skip,
-                     "The old node does not have improvements to "
-                     "rabbit_amqqueue_process and rabbit_node_monitor"}
-            end;
-        _ ->
-            %% Other testcases or failure to setup broker and client.
-            Config2
-    end.
+    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,
diff --git a/deps/rabbit/test/direct_reply_to_amqp_SUITE.erl b/deps/rabbit/test/direct_reply_to_amqp_SUITE.erl
@@ -70,16 +70,10 @@ init_per_group(Group, Config) ->
     Config1 = rabbit_ct_helpers:set_config(
                 Config, [{rmq_nodes_count, Nodes},
                          {rmq_nodename_suffix, Suffix}]),
-    Config2 = rabbit_ct_helpers:run_setup_steps(
-                Config1,
-                rabbit_ct_broker_helpers:setup_steps() ++
-                rabbit_ct_client_helpers:setup_steps()),
-    case rabbit_ct_broker_helpers:enable_feature_flag(Config2, 'rabbitmq_4.2.0') of
-        ok ->
-            Config2;
-        {skip, _} = Skip ->
-            Skip
-    end.
+    rabbit_ct_helpers:run_setup_steps(
+      Config1,
+      rabbit_ct_broker_helpers:setup_steps() ++
+      rabbit_ct_client_helpers:setup_steps()).
 
 end_per_group(_Group, Config) ->
     rabbit_ct_helpers:run_teardown_steps(
diff --git a/deps/rabbit/test/direct_reply_to_amqpl_SUITE.erl b/deps/rabbit/test/direct_reply_to_amqpl_SUITE.erl
diff --git a/deps/rabbit/test/msg_size_metrics_SUITE.erl b/deps/rabbit/test/msg_size_metrics_SUITE.erl
diff --git a/deps/rabbit/test/rabbit_direct_reply_to_prop_SUITE.erl b/deps/rabbit/test/rabbit_direct_reply_to_prop_SUITE.erl
