Remove feature flag 'rabbitmq_4.1.0' compat code

(cherry picked from commit 34df56a)

Author: ansd

deps/rabbit/src/rabbit_amqp_reader.erl

@@ -219,9 +219,6 @@ handle_other({rabbit_call, From, {info, Items}}, State) ->
             end,
     gen_server:reply(From, Reply),
     State;
-handle_other({'$gen_call', From, Req}, State) ->
-    %% Delete this function clause when feature flag 'rabbitmq_4.1.0' becomes required.
-    handle_other({rabbit_call, From, Req}, State);
 handle_other({'$gen_cast', {force_event_refresh, Ref}}, State) ->
     case ?IS_RUNNING(State) of
         true ->

deps/rabbit/src/rabbit_reader.erl

@@ -662,9 +662,6 @@ handle_other({rabbit_call, From, {info, Items}}, State) ->
                            catch Error -> {error, Error}
                            end),
     State;
-handle_other({'$gen_call', From, Req}, State) ->
-    %% Delete this function clause when feature flag 'rabbitmq_4.1.0' becomes required.
-    handle_other({rabbit_call, From, Req}, State);
 handle_other({'$gen_cast', {force_event_refresh, Ref}}, State)
   when ?IS_RUNNING(State) ->
     rabbit_event:notify(
@@ -1886,17 +1883,12 @@ connection_duration(ConnectedAt) ->
     end.
 
 gen_call(Pid, Req, Timeout) ->
-    case rabbit_feature_flags:is_enabled('rabbitmq_4.1.0') of
-        true ->
-            %% We use gen:call/4 with label rabbit_call instead of gen_server:call/3 with label '$gen_call'
-            %% because cowboy_websocket does not let rabbit_web_amqp_handler handle '$gen_call' messages:
-            %% https://github.com/ninenines/cowboy/blob/2.12.0/src/cowboy_websocket.erl#L427-L430
-            case catch gen:call(Pid, rabbit_call, Req, Timeout) of
-                {ok, Res} ->
-                    Res;
-                {'EXIT', Reason} ->
-                    exit({Reason, {?MODULE, ?FUNCTION_NAME, [Pid, Req, Timeout]}})
-            end;
-        false ->
-            gen_server:call(Pid, Req, Timeout)
+    %% We use gen:call/4 with label rabbit_call instead of gen_server:call/3 with label '$gen_call'
+    %% because cowboy_websocket does not let rabbit_web_amqp_handler handle '$gen_call' messages:
+    %% https://github.com/ninenines/cowboy/blob/2.12.0/src/cowboy_websocket.erl#L427-L430
+    case catch gen:call(Pid, rabbit_call, Req, Timeout) of
+        {ok, Res} ->
+            Res;
+        {'EXIT', Reason} ->
+            exit({Reason, {?MODULE, ?FUNCTION_NAME, [Pid, Req, Timeout]}})
     end.

deps/rabbit/test/metrics_SUITE.erl

@@ -143,12 +143,8 @@ connection_metric_idemp(Config, {N, R}) ->
                                5000),
     Table2 = [ Pid || {Pid, _} <- read_table_rpc(Config, connection_coarse_metrics)],
     % refresh stats 'R' times
-    [[begin
-          Pid ! emit_stats
-      end|| Pid <- Table] || _ <- lists:seq(1, R)],
-    [begin
-         _ = gen_server:call(Pid, {info, [pid]})
-     end|| Pid <- Table],
+    [[Pid ! emit_stats || Pid <- Table] || _ <- lists:seq(1, R)],
+    [_ = gen:call(Pid, rabbit_call, {info, [pid]}, 9000) || Pid <- Table],
     force_metric_gc(Config),
     TableAfter = [ Pid || {Pid, _} <- read_table_rpc(Config, connection_metrics)],
     TableAfter2 = [ Pid || {Pid, _} <- read_table_rpc(Config, connection_coarse_metrics)],

deps/rabbitmq_mqtt/src/rabbit_mqtt_processor.erl

@@ -214,9 +214,6 @@ process_connect(
         {TraceState, ConnName} = init_trace(VHost, ConnName0),
         ok = rabbit_mqtt_keepalive:start(KeepaliveSecs, Socket),
         Exchange = rabbit_misc:r(VHost, exchange, persistent_term:get(?PERSISTENT_TERM_EXCHANGE)),
-        %% To simplify logic, we decide at connection establishment time to stick
-        %% with either binding args v1 or v2 for the lifetime of the connection.
-        BindingArgsV2 = rabbit_feature_flags:is_enabled('rabbitmq_4.1.0'),
         ProtoVerAtom = proto_integer_to_atom(ProtoVer),
         MsgIcptCtx = #{protocol => ProtoVerAtom,
                        vhost => VHost,
@@ -246,7 +243,6 @@ process_connect(
                           will_msg = WillMsg,
                           max_packet_size_outbound = MaxPacketSize,
                           topic_alias_maximum_outbound = TopicAliasMaxOutbound,
-                          binding_args_v2 = BindingArgsV2,
                           msg_interceptor_ctx = MsgIcptCtx},
                auth_state = #auth_state{
                                user = User,
@@ -459,8 +455,7 @@ process_request(?SUBSCRIBE,
                                  packet_id  = SubscribePktId,
                                  subscriptions = Subscriptions},
                    payload = undefined},
-                State0 = #state{cfg = #cfg{proto_ver = ProtoVer,
-                                           binding_args_v2 = BindingArgsV2}}) ->
+                State0 = #state{cfg = #cfg{proto_ver = ProtoVer}}) ->
     ?LOG_DEBUG("Received a SUBSCRIBE with subscription(s) ~p", [Subscriptions]),
     DisconnectOnUnauthorized = rabbit_mqtt_util:env(disconnect_on_unauthorized),
     {ResultRev, RetainedRev, State1} =
@@ -491,8 +486,7 @@ process_request(?SUBSCRIBE,
                       maybe
                           {ok, Q} ?= ensure_queue(QoS, S0),
                           QName = amqqueue:get_name(Q),
-                          BindingArgs = binding_args_for_proto_ver(ProtoVer, TopicFilter,
-                                                                   Opts, BindingArgsV2),
+                          BindingArgs = binding_args_for_proto_ver(ProtoVer, TopicFilter, Opts),
                           ok ?= add_subscription(TopicFilter, BindingArgs, QName, S0),
                           ok ?= maybe_delete_old_subscription(TopicFilter, Opts, S0),
                           Subs = maps:put(TopicFilter, Opts, S0#state.subscriptions),
@@ -541,11 +535,10 @@ process_request(?UNSUBSCRIBE,
     {ReasonCodes, State} =
     lists:foldl(
       fun(TopicFilter, {L, #state{subscriptions = Subs0,
-                                  cfg = #cfg{proto_ver = ProtoVer,
-                                             binding_args_v2 = BindingArgsV2}} = S0}) ->
+                                  cfg = #cfg{proto_ver = ProtoVer}} = S0}) ->
               case maps:take(TopicFilter, Subs0) of
                   {Opts, Subs} ->
-                      BindingArgs = binding_args_for_proto_ver(ProtoVer, TopicFilter, Opts, BindingArgsV2),
+                      BindingArgs = binding_args_for_proto_ver(ProtoVer, TopicFilter, Opts),
                       case delete_subscription(
                              TopicFilter, BindingArgs, Opts#mqtt_subscription_opts.qos, S0) of
                           ok ->
@@ -913,8 +906,7 @@ init_subscriptions(_, State) ->
 -spec init_subscriptions0(qos(), state()) ->
     {ok, subscriptions()} | {error, reason_code()}.
 init_subscriptions0(QoS, State = #state{cfg = #cfg{proto_ver = ProtoVer,
-                                                   exchange = Exchange,
-                                                   binding_args_v2 = BindingArgsV2}}) ->
+                                                   exchange = Exchange }}) ->
     Bindings =
     rabbit_binding:list_for_source_and_destination(
       Exchange,
@@ -934,7 +926,7 @@ init_subscriptions0(QoS, State = #state{cfg = #cfg{proto_ver = ProtoVer,
                          case ProtoVer of
                              ?MQTT_PROTO_V5 ->
                                  %% session upgrade
-                                 NewBindingArgs = binding_args_for_proto_ver(ProtoVer, TopicFilter, Opts, BindingArgsV2),
+                                 NewBindingArgs = binding_args_for_proto_ver(ProtoVer, TopicFilter, Opts),
                                  ok = recreate_subscription(TopicFilter, Args, NewBindingArgs, QoS, State);
                              _ ->
                                  ok
@@ -953,19 +945,13 @@ init_subscriptions0(QoS, State = #state{cfg = #cfg{proto_ver = ProtoVer,
                                                 %% binding args v1
                                                 Opts0 = #mqtt_subscription_opts{} = lists:keyfind(
                                                                                       mqtt_subscription_opts, 1, Args),
-                                                case BindingArgsV2 of
-                                                    true ->
-                                                        %% Migrate v1 to v2.
-                                                        %% Note that this migration must be in place even for some versions
-                                                        %% (jump upgrade) after feature flag 'rabbitmq_4.1.0' has become
-                                                        %% required since enabling the feature flag doesn't migrate binding
-                                                        %% args for existing connections.
-                                                        NewArgs = binding_args_for_proto_ver(
-                                                                    ProtoVer, TopicFilter, Opts0, BindingArgsV2),
-                                                        ok = recreate_subscription(TopicFilter, Args, NewArgs, QoS, State);
-                                                    false ->
-                                                        ok
-                                                end,
+                                                %% Migrate v1 to v2.
+                                                %% Note that this migration must be in place even for some versions
+                                                %% (jump upgrade) after feature flag 'rabbitmq_4.1.0' has become
+                                                %% required since enabling the feature flag doesn't migrate binding
+                                                %% args for existing connections.
+                                                NewArgs = binding_args_for_proto_ver(ProtoVer, TopicFilter, Opts0),
+                                                ok = recreate_subscription(TopicFilter, Args, NewArgs, QoS, State),
                                                 Opts0
                                         end;
                                     _ ->
@@ -1000,10 +986,11 @@ recreate_subscription(TopicFilter, OldBindingArgs, NewBindingArgs, Qos, State) -
 -spec hand_off_to_retainer(pid(), topic(), mqtt_msg()) -> ok.
 hand_off_to_retainer(RetainerPid, Topic0, Msg = #mqtt_msg{payload = Payload}) ->
     Topic = amqp_to_mqtt(Topic0),
-    if Payload =:= <<>> ->
-           rabbit_mqtt_retainer:clear(RetainerPid, Topic);
-       true ->
-           rabbit_mqtt_retainer:retain(RetainerPid, Topic, Msg)
+    case Payload of
+        <<>> ->
+            rabbit_mqtt_retainer:clear(RetainerPid, Topic);
+        _ ->
+            rabbit_mqtt_retainer:retain(RetainerPid, Topic, Msg)
     end.
 
 -spec send_retained_messages([{topic_filter(), qos()}], state()) -> state().
@@ -1538,18 +1525,13 @@ consume(Q, QoS, #state{
             Err
     end.
 
-binding_args_for_proto_ver(?MQTT_PROTO_V3, _, _, _) ->
+binding_args_for_proto_ver(?MQTT_PROTO_V3, _, _) ->
     [];
-binding_args_for_proto_ver(?MQTT_PROTO_V4, _, _, _) ->
+binding_args_for_proto_ver(?MQTT_PROTO_V4, _, _) ->
     [];
-binding_args_for_proto_ver(?MQTT_PROTO_V5, TopicFilter, SubOpts0, V2) ->
-    SubOpts = case V2 of
-                  true ->
-                      Table = subscription_opts_to_table(SubOpts0),
-                      {<<"x-mqtt-subscription-opts">>, table, Table};
-                  false ->
-                      SubOpts0
-              end,
+binding_args_for_proto_ver(?MQTT_PROTO_V5, TopicFilter, SubOpts0) ->
+    Table = subscription_opts_to_table(SubOpts0),
+    SubOpts = {<<"x-mqtt-subscription-opts">>, table, Table},
     BindingKey = mqtt_to_amqp(TopicFilter),
     [SubOpts, {<<"x-binding-key">>, longstr, BindingKey}].
 
@@ -1600,13 +1582,12 @@ delete_subscription(TopicFilter, BindingArgs, Qos, State) ->
 %% Subscription will be identical to that in the previous Subscription, although its
 %% Subscription Options could be different." [v5 3.8.4]
 maybe_delete_old_subscription(TopicFilter, Opts, State = #state{subscriptions = Subs,
-                                                                cfg = #cfg{proto_ver = ProtoVer,
-                                                                           binding_args_v2 = BindingArgsV2}}) ->
+                                                                cfg = #cfg{proto_ver = ProtoVer}}) ->
     case Subs of
         #{TopicFilter := OldOpts}
           when OldOpts =/= Opts ->
             delete_subscription(TopicFilter,
-                                binding_args_for_proto_ver(ProtoVer, TopicFilter, OldOpts, BindingArgsV2),
+                                binding_args_for_proto_ver(ProtoVer, TopicFilter, OldOpts),
                                 OldOpts#mqtt_subscription_opts.qos,
                                 State);
         _ ->

deps/rabbitmq_mqtt/test/federation_SUITE.erl

@@ -40,12 +40,7 @@ end_per_suite(Config) ->
       rabbit_ct_broker_helpers:teardown_steps()).
 
 init_per_testcase(Testcase, Config) ->
-    case rabbit_ct_broker_helpers:enable_feature_flag(Config, 'rabbitmq_4.1.0') of
-        ok ->
-            rabbit_ct_helpers:testcase_started(Config, Testcase);
-        Skip = {skip, _Reason} ->
-            Skip
-    end.
+    rabbit_ct_helpers:testcase_started(Config, Testcase).
 
 end_per_testcase(Testcase, Config) ->
     rabbit_ct_helpers:testcase_finished(Config, Testcase).