For 4.0: Require all stable feature flags added up to 3.13.0

deps/rabbit/src/rabbit_core_ff.erl

@@ -101,22 +101,22 @@
    {restart_streams,
     #{desc          => "Support for restarting streams with optional preferred next leader argument."
       "Used to implement stream leader rebalancing",
-      stability     => stable,
+      stability     => required,
       depends_on    => [stream_queue]
      }}).
 
 -rabbit_feature_flag(
    {stream_sac_coordinator_unblock_group,
     #{desc          => "Bug fix to unblock a group of consumers in a super stream partition",
       doc_url       => "https://github.com/rabbitmq/rabbitmq-server/issues/7743",
-      stability     => stable,
+      stability     => required,
       depends_on    => [stream_single_active_consumer]
      }}).
 
 -rabbit_feature_flag(
    {stream_filtering,
     #{desc          => "Support for stream filtering.",
-      stability     => stable,
+      stability     => required,
       depends_on    => [stream_queue]
      }}).
 
@@ -153,7 +153,7 @@
    {stream_update_config_command,
     #{desc          => "A new internal command that is used to update streams as "
                         "part of a policy.",
-      stability     => stable,
+      stability     => required,
       depends_on    => [stream_queue]
      }}).
 

deps/rabbit/src/rabbit_stream_coordinator.erl

@@ -174,19 +174,14 @@ restart_stream(QRes, Options)
 restart_stream(Q, Options)
   when ?is_amqqueue(Q) andalso
        ?amqqueue_is_stream(Q) ->
-    case rabbit_feature_flags:is_enabled(restart_streams) of
-        true ->
-            rabbit_log:info("restarting stream ~s in vhost ~s with options ~p",
-                            [maps:get(name, amqqueue:get_type_state(Q)), amqqueue:get_vhost(Q), Options]),
-            #{name := StreamId} = amqqueue:get_type_state(Q),
-            case process_command({restart_stream, StreamId, Options}) of
-                {ok, {ok, LeaderPid}, _} ->
-                    {ok, node(LeaderPid)};
-                Err ->
-                    Err
-            end;
-        false ->
-            {error, {feature_flag_not_enabled, restart_stream}}
+    rabbit_log:info("restarting stream ~s in vhost ~s with options ~p",
+                    [maps:get(name, amqqueue:get_type_state(Q)), amqqueue:get_vhost(Q), Options]),
+    #{name := StreamId} = amqqueue:get_type_state(Q),
+    case process_command({restart_stream, StreamId, Options}) of
+        {ok, {ok, LeaderPid}, _} ->
+            {ok, node(LeaderPid)};
+        Err ->
+            Err
     end.
 
 delete_stream(Q, ActingUser)
@@ -254,22 +249,17 @@ policy_changed(Q) when ?is_amqqueue(Q) ->
     {ok, ok, ra:server_id()} | {error, not_supported | term()}.
 update_config(Q, Config)
   when ?is_amqqueue(Q) andalso is_map(Config) ->
-    case rabbit_feature_flags:is_enabled(stream_update_config_command) of
-        true ->
-            %% there are the only a few configuration keys that are safe to
-            %% update
-            StreamId = maps:get(name, amqqueue:get_type_state(Q)),
-            case maps:with([filter_size,
-                            retention,
-                            writer_mod,
-                            replica_mod], Config) of
-                Conf when map_size(Conf) > 0 ->
-                    process_command({update_config, StreamId, Conf});
-                _ ->
-                    {error, no_updatable_keys}
-            end;
-        false ->
-            {error, feature_not_enabled}
+    %% there are the only a few configuration keys that are safe to
+    %% update
+    StreamId = maps:get(name, amqqueue:get_type_state(Q)),
+    case maps:with([filter_size,
+                    retention,
+                    writer_mod,
+                    replica_mod], Config) of
+        Conf when map_size(Conf) > 0 ->
+            process_command({update_config, StreamId, Conf});
+        _ ->
+            {error, no_updatable_keys}
     end.
 
 sac_state(#?MODULE{single_active_consumer = SacState}) ->

deps/rabbit/src/rabbit_stream_queue.erl

@@ -96,8 +96,7 @@
                         soft_limit :: non_neg_integer(),
                         slow = false :: boolean(),
                         readers = #{} :: #{rabbit_types:ctag() => #stream{}},
-                        writer_id :: binary(),
-                        filtering_supported :: boolean()
+                        writer_id :: binary()
                        }).
 
 -import(rabbit_queue_type_util, [args_policy_lookup/3]).
@@ -286,8 +285,7 @@ consume(Q, #{no_ack := true,
 consume(Q, #{limiter_active := true}, _State)
   when ?amqqueue_is_stream(Q) ->
     {error, global_qos_not_supported_for_queue_type};
-consume(Q, Spec,
-        #stream_client{filtering_supported = FilteringSupported} = QState0)
+consume(Q, Spec, #stream_client{} = QState0)
   when ?amqqueue_is_stream(Q) ->
     %% Messages should include the offset as a custom header.
     case get_local_pid(QState0) of
@@ -307,26 +305,19 @@ consume(Q, Spec,
                 {error, _} = Err ->
                     Err;
                 {ok, OffsetSpec} ->
-                    FilterSpec = filter_spec(Args),
-                    case {FilterSpec, FilteringSupported} of
-                        {#{filter_spec := _}, false} ->
-                            {protocol_error, precondition_failed,
-                             "Filtering is not supported", []};
-                        _ ->
-                            ConsumerPrefetchCount = case Mode of
-                                                        {simple_prefetch, C} -> C;
-                                                        _ -> 0
-                                                    end,
-                            AckRequired = not NoAck,
-                            rabbit_core_metrics:consumer_created(
-                              ChPid, ConsumerTag, ExclusiveConsume, AckRequired,
-                              QName, ConsumerPrefetchCount, false, up, Args),
-                            %% reply needs to be sent before the stream
-                            %% begins sending
-                            maybe_send_reply(ChPid, OkMsg),
-                            _ = rabbit_stream_coordinator:register_local_member_listener(Q),
-                            begin_stream(QState, ConsumerTag, OffsetSpec, Mode, AckRequired, FilterSpec)
-                    end
+                    ConsumerPrefetchCount = case Mode of
+                                                {simple_prefetch, C} -> C;
+                                                _ -> 0
+                                            end,
+                    AckRequired = not NoAck,
+                    rabbit_core_metrics:consumer_created(
+                      ChPid, ConsumerTag, ExclusiveConsume, AckRequired,
+                      QName, ConsumerPrefetchCount, false, up, Args),
+                    %% reply needs to be sent before the stream
+                    %% begins sending
+                    maybe_send_reply(ChPid, OkMsg),
+                    _ = rabbit_stream_coordinator:register_local_member_listener(Q),
+                    begin_stream(QState, ConsumerTag, OffsetSpec, Mode, AckRequired, filter_spec(Args))
             end;
         {undefined, _} ->
             {protocol_error, precondition_failed,
@@ -510,8 +501,7 @@ deliver(QSs, Msg, Options) ->
     lists:foldl(
       fun({Q, stateless}, {Qs, Actions}) ->
               LeaderPid = amqqueue:get_pid(Q),
-              ok = osiris:write(LeaderPid,
-                                stream_message(Msg, filtering_supported())),
+              ok = osiris:write(LeaderPid, stream_message(Msg)),
               {Qs, Actions};
          ({Q, S0}, {Qs, Actions0}) ->
               {S, Actions} = deliver0(maps:get(correlation, Options, undefined),
@@ -526,11 +516,9 @@ deliver0(MsgId, Msg,
                         next_seq = Seq,
                         correlation = Correlation0,
                         soft_limit = SftLmt,
-                        slow = Slow0,
-                        filtering_supported = FilteringSupported} = State,
+                        slow = Slow0} = State,
          Actions0) ->
-    ok = osiris:write(LeaderPid, WriterId, Seq,
-                      stream_message(Msg, FilteringSupported)),
+    ok = osiris:write(LeaderPid, WriterId, Seq, stream_message(Msg)),
     Correlation = case MsgId of
                       undefined ->
                           Correlation0;
@@ -547,19 +535,14 @@ deliver0(MsgId, Msg,
                          correlation = Correlation,
                          slow = Slow}, Actions}.
 
-stream_message(Msg, FilteringSupported) ->
+stream_message(Msg) ->
     McAmqp = mc:convert(mc_amqp, Msg),
     MsgData = mc:protocol_state(McAmqp),
-    case FilteringSupported of
-        true ->
-            case mc:x_header(<<"x-stream-filter-value">>, McAmqp) of
-                undefined ->
-                    MsgData;
-                {utf8, Value} ->
-                    {Value, MsgData}
-            end;
-        false ->
-            MsgData
+    case mc:x_header(<<"x-stream-filter-value">>, McAmqp) of
+        undefined ->
+            MsgData;
+        {utf8, Value} ->
+            {Value, MsgData}
     end.
 
 -spec dequeue(_, _, _, _, client()) -> no_return().
@@ -936,8 +919,7 @@ init(Q) when ?is_amqqueue(Q) ->
                                 name = amqqueue:get_name(Q),
                                 leader = Leader,
                                 writer_id = WriterId,
-                                soft_limit = SoftLimit,
-                                filtering_supported = filtering_supported()}};
+                                soft_limit = SoftLimit}};
         {ok, stream_not_found, _} ->
             {error, stream_not_found};
         {error, coordinator_unavailable} = E ->
@@ -1294,8 +1276,7 @@ notify_decorators(Q) when ?is_amqqueue(Q) ->
 
 resend_all(#stream_client{leader = LeaderPid,
                           writer_id = WriterId,
-                          correlation = Corrs,
-                          filtering_supported = FilteringSupported} = State) ->
+                          correlation = Corrs} = State) ->
     Msgs = lists:sort(maps:values(Corrs)),
     case Msgs of
         [] -> ok;
@@ -1304,8 +1285,7 @@ resend_all(#stream_client{leader = LeaderPid,
                              [Seq, maps:size(Corrs)])
     end,
     [begin
-         ok = osiris:write(LeaderPid, WriterId, Seq,
-                           stream_message(Msg, FilteringSupported))
+         ok = osiris:write(LeaderPid, WriterId, Seq, stream_message(Msg))
      end || {Seq, Msg} <- Msgs],
     State.
 
@@ -1340,9 +1320,6 @@ list_with_minimum_quorum() ->
 
 is_stateful() -> true.
 
-filtering_supported() ->
-    rabbit_feature_flags:is_enabled(stream_filtering).
-
 get_nodes(Q) when ?is_amqqueue(Q) ->
     #{nodes := Nodes} = amqqueue:get_type_state(Q),
     Nodes.

deps/rabbit/src/rabbit_stream_sac_coordinator.erl

@@ -629,32 +629,12 @@ handle_consumer_removal(Group0, Consumer, Stream, ConsumerName) ->
             end
     end.
 
-message_type() ->
-    case has_unblock_group_support() of
-        true ->
-            map;
-        false ->
-            tuple
-    end.
-
 notify_consumer_effect(Pid, SubId, Stream, Name, Active) ->
     notify_consumer_effect(Pid, SubId, Stream, Name, Active, false).
 
 notify_consumer_effect(Pid, SubId, Stream, Name, Active, SteppingDown) ->
-    notify_consumer_effect(Pid, SubId, Stream, Name, Active, SteppingDown, message_type()).
+    notify_consumer_effect(Pid, SubId, Stream, Name, Active, SteppingDown, map).
 
-notify_consumer_effect(Pid, SubId, _Stream, _Name, Active, false = _SteppingDown, tuple) ->
-    mod_call_effect(Pid,
-                    {sac,
-                     {{subscription_id, SubId},
-                      {active, Active},
-                      {extra, []}}});
-notify_consumer_effect(Pid, SubId, _Stream, _Name, Active, true = _SteppingDown, tuple) ->
-    mod_call_effect(Pid,
-                    {sac,
-                     {{subscription_id, SubId},
-                      {active, Active},
-                      {extra, [{stepping_down, true}]}}});
 notify_consumer_effect(Pid, SubId, Stream, Name, Active, false = _SteppingDown, map) ->
     mod_call_effect(Pid,
                     {sac, #{subscription_id => SubId,
@@ -776,6 +756,3 @@ mod_call_effect(Pid, Msg) ->
 send_message(ConnectionPid, Msg) ->
     ConnectionPid ! Msg,
     ok.
-
-has_unblock_group_support() ->
-    rabbit_feature_flags:is_enabled(stream_sac_coordinator_unblock_group).

deps/rabbit/test/amqp_client_SUITE.erl

@@ -3138,7 +3138,6 @@ global_counters(Config) ->
     ok = amqp10_client:close_connection(Connection).
 
 stream_filtering(Config) ->
-    ok = rabbit_ct_broker_helpers:enable_feature_flag(Config, ?FUNCTION_NAME),
     Stream = atom_to_binary(?FUNCTION_NAME),
     Address = rabbitmq_amqp_address:queue(Stream),
     Ch = rabbit_ct_client_helpers:open_channel(Config),