Deprecated features: New module to manage deprecated features (!)introduce-deprecation-feature-flags

This introduces a way to declare deprecated features in the code, not
only in our communication. The new module allows to disallow the use of
a deprecated feature and/or warn the user when he relies on such a
feature.

[Why]
Currently, we only tell people about deprecated features through blog
posts and the mailing-list. This might be insufficiant for our users
that a feature they use will be removed in a future version:
* They may not read our blog or mailing-list
* They may not understand that they use such a deprecated feature
* They might wait for the big removal before they plan testing
* They might not take it seriously enough

The idea behind this patch is to increase the chance that users notice
that they are using something which is about to be dropped from
RabbitMQ. Anopther benefit is that they should be able to test how
RabbitMQ will behave in the future before the actual removal. This
should allow them to test and plan changes.

[How]
When a feature is deprecated in other large projects (such as FreeBSD
where I took the idea from), it goes through a lifecycle:
1. The feature is still available, but users get a warning somehow when
   they use it. They can disable it to test.
2. The feature is still available, but disabled out-of-the-box. Users
   can re-enable it (and get a warning).
3. The feature is disconnected from the build. Therefore, the code
   behind it is still there, but users have to recompile the thing to be
   able to use it.
4. The feature is removed from the source code. Users have to adapt or
   they can't upgrade anymore.

The solution in this patch offers the same lifecycle. A deprecated
feature will be in one of these deprecation phases:
1. `permitted_by_default`: The feature is available. Users get a warning
   if they use it. They can disable it from the configuration.
2. `denied_by_default`: The feature is available but disabled by
   default. Users get an error if they use it and RabbitMQ behaves like
   the feature is removed. They can re-enable is from the configuration
   and get a warning.
3. `disconnected`: The feature is present in the source code, but is
   disabled and can't be re-enabled without recompiling RabbitMQ. Users
   get the same behavior as if the code was removed.
4. `removed`: The feature's code is gone.

The whole thing is based on the feature flags subsystem, but it has the
following differences with other feature flags:
* The semantic is reversed: the feature flag behind a deprecated feature
  is disabled when the deprecated feature is permitted, or enabled when
  the deprecated feature is denied.
* The feature flag behind a deprecated feature is enabled out-of-the-box
  (meaning the deprecated feature is denied):
    * if the deprecation phase is `permitted_by_default` and the
      configuration denies the deprecated feature
    * if the deprecation phase is `denied_by_default` and the
      configuration doesn't permit the deprecated feature
    * if the deprecation phase is `disconnected` or `removed`
* Feature flags behind deprecated feature don't appear in feature flags
  listings.

Otherwise, deprecated features' feature flags are managed like other
feature flags, in particular inside clusters.

To declare a deprecated feature:

    -rabbit_deprecated_feature(
       {my_deprecated_feature,
        #{deprecation_phase => permitted_by_default,
          msgs => #{when_permitted => "This feature will be removed in RabbitMQ X.0"},
         }}).

Then, to check the state of a deprecated feature in the code:

    case rabbit_deprecated_features:is_permitted(my_deprecated_feature) of
        true ->
            %% The deprecated feature is still permitted.
            ok;
        false ->
            %% The deprecated feature is gone or should be considered
            %% unavailable.
            error
    end.

Warnings and errors are logged automatically. A message is generated
automatically, but it is possible to define a message in the deprecated
feature flag declaration like in the example above.

Here is an example of a logged warning that was generated automatically:

    Feature `my_deprecated_feature` is deprecated.
    By default, this feature can still be used for now.
    Its use will not be permitted by default in a future minor RabbitMQ version and the feature will be removed from a future major RabbitMQ version; actual versions to be determined.
    To continue using this feature when it is not permitted by default, set the following parameter in your configuration:
        "deprecated_features.permit.my_deprecated_feature = true"
    To test RabbitMQ as if the feature was removed, set this in your configuration:
        "deprecated_features.permit.my_deprecated_feature = false"

To override the default state of `permitted_by_default` and
`denied_by_default` deprecation phases, users can set the following
configuration:

    # In rabbitmq.conf:
    deprecated_features.permit.my_deprecated_feature = true # or false

The actual behavior protected by a deprecated feature check is out of
scope for this subsystem. It is the repsonsibility of each deprecated
feature code to determine what to do when the deprecated feature is
denied.

V1: Deprecated feature states are initially computed during the
    initialization of the registry, based on their deprecation phase and
    possibly the configuration. They don't go through the `enable/1`
    code at all.

V2: Manage deprecated feature states as any other non-required
    feature flags. This allows to execute an `is_feature_used()`
    callback to determine if a deprecated feature can be denied. This
    also allows to prevent the RabbitMQ node from starting if it
    continues to use a deprecated feature.

V3: Manage deprecated feature states from the registry initialization
    again. This is required because we need to know very early if some
    of them are denied, so that an upgrade to a version of RabbitMQ
    where a deprecated feature is disconnected or removed can be
    performed.

    To still prevent the start of a RabbitMQ node when a denied
    deprecated feature is actively used, we run the `is_feature_used()`
    callback of all denied deprecated features as part of the
    `sync_cluster()` task. This task is executed as part of a feature
    flag refresh executed when RabbitMQ starts or when plugins are
    enabled. So even though a deprecated feature is marked as denied in
    the registry early in the boot process, we will still abort the
    start of a RabbitMQ node if the feature is used.

V4: Support context-dependent warnings. It is now possible to set a
    specific message when deprecated feature is permitted, when it is
    denied and when it is removed. Generic per-context messages are
    still generated.

V5: Improve default warning messages, thanks to @pstack2021.

V6: Rename the configuration variable from `permit_deprecated_features.*`
    to `deprecated_features.permit.*`. As @michaelklishin said, we tend
    to use shorter top-level names.

14 files changed, 1779 insertions, 195 deletions

diff --git a/deps/rabbit/BUILD.bazel b/deps/rabbit/BUILD.bazel
index 49323983e0..302667077e 100644
--- a/deps/rabbit/BUILD.bazel
+++ b/deps/rabbit/BUILD.bazel
@@ -380,6 +380,14 @@ rabbitmq_integration_suite(
 )
 
 rabbitmq_integration_suite(
+    name = "deprecated_features_SUITE",
+    size = "medium",
+    additional_beam = [
+        ":feature_flags_v2_SUITE_beam_files",
+    ],
+)
+
+rabbitmq_integration_suite(
     name = "disconnect_detected_during_alarm_SUITE",
     size = "medium",
 )
diff --git a/deps/rabbit/Makefile b/deps/rabbit/Makefile
index ea6704b6e4..20aa7d848e 100644
--- a/deps/rabbit/Makefile
+++ b/deps/rabbit/Makefile
@@ -241,7 +241,7 @@ ct-slow: CT_SUITES = $(SLOW_CT_SUITES)
 # --------------------------------------------------------------------
 
 RMQ_ERLC_OPTS += -I $(DEPS_DIR)/rabbit_common/include
-EDOC_OPTS += {preprocess,true}
+EDOC_OPTS += {preprocess,true},{includes,["."]}
 
 ifdef INSTRUMENT_FOR_QC
 RMQ_ERLC_OPTS += -DINSTR_MOD=gm_qc
diff --git a/deps/rabbit/docs/rabbitmq.conf.example b/deps/rabbit/docs/rabbitmq.conf.example
index 00dec1473c..43e21dac78 100644
--- a/deps/rabbit/docs/rabbitmq.conf.example
+++ b/deps/rabbit/docs/rabbitmq.conf.example
@@ -548,6 +548,25 @@
 ## NB: Change these only if you understand what you are doing!
 ##
 
+## To permit or deny a deprecated feature when it is in its
+## `permitted_by_default` or `denied_by_default` deprecation phase, the
+## default state can be overriden from the configuration.
+##
+## When a deprecated feature is permitted by default (first phase of the
+## deprecation period), it means the feature is available by default and can
+## be turned off by setting it to false in the configuration.
+##
+## When a deprecated feature is denied by default (second phase of the
+## deprecation period), it means the feature is unavailable by default but can
+## be turned back on by setting it to true in the configuration.
+##
+## When a deprecated feature is "disconnected" or "removed" (last two phases
+## of the deprecation period), it is no longer possible to turn it back on
+## from the configuration.
+##
+# deprecated_features.permit.a_deprecated_feature = true
+# deprecated_features.permit.another_deprecated_feature = false
+
 ## Timeout used when waiting for Mnesia tables in a cluster to
 ## become available.
 ##
diff --git a/deps/rabbit/priv/schema/rabbit.schema b/deps/rabbit/priv/schema/rabbit.schema
index 07643b050a..0ffe00c692 100644
--- a/deps/rabbit/priv/schema/rabbit.schema
+++ b/deps/rabbit/priv/schema/rabbit.schema
@@ -2120,18 +2120,17 @@ end}.
 %% =====================================
 %%
 
-%% NOTE: `true` is intentionally omitted - add it back when mirrored
-%% queue deprecation is converted to use deprecated features system.
 {mapping,
- "deprecated_features.permit.$name", "rabbit.permitted_deprecated_features",
-    [{datatype, {enum, [false]}}]
+ "deprecated_features.permit.$name", "rabbit.permit_deprecated_features",
+ [{datatype, {enum, [true, false]}}]
 }.
 
 %% This converts:
-%%     deprecated_features.permit.my_feature = false
+%%     deprecated_features.permit.my_feature = true
 %% to:
-%%     {rabbit, [{permitted_deprecated_features, #{my_feature => false}}]}.
-{translation, "rabbit.permitted_deprecated_features",
+%%     {rabbit, [{permit_deprecated_features, #{my_feature => true}}]}.
+
+{translation, "rabbit.permit_deprecated_features",
  fun(Conf) ->
          Settings = cuttlefish_variable:filter_by_prefix(
                       "deprecated_features.permit", Conf),
diff --git a/deps/rabbit/src/rabbit.erl b/deps/rabbit/src/rabbit.erl
index 3e6e705a55..5e0c9cee64 100644
--- a/deps/rabbit/src/rabbit.erl
+++ b/deps/rabbit/src/rabbit.erl
@@ -512,7 +512,10 @@ start_apps(Apps, RestartTypes) ->
     %% We need to load all applications involved in order to be able to
     %% find new feature flags.
     app_utils:load_applications(Apps),
-    ok = rabbit_feature_flags:refresh_feature_flags_after_app_load(),
+    case rabbit_feature_flags:refresh_feature_flags_after_app_load() of
+        ok    -> ok;
+        Error -> throw(Error)
+    end,
     rabbit_prelaunch_conf:decrypt_config(Apps),
     lists:foreach(
       fun(App) ->
@@ -932,7 +935,10 @@ start(normal, []) ->
         %% once, because it does not involve running code from the
         %% plugins.
         ok = app_utils:load_applications(Plugins),
-        ok = rabbit_feature_flags:refresh_feature_flags_after_app_load(),
+        case rabbit_feature_flags:refresh_feature_flags_after_app_load() of
+            ok     -> ok;
+            Error1 -> throw(Error1)
+        end,
 
         persist_static_configuration(),
 
diff --git a/deps/rabbit/src/rabbit_deprecated_features.erl b/deps/rabbit/src/rabbit_deprecated_features.erl
new file mode 100644
index 0000000000..ec48ba59ca
--- /dev/null
+++ b/deps/rabbit/src/rabbit_deprecated_features.erl
@@ -0,0 +1,599 @@
+%% 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.
+%%
+
+%% @author The RabbitMQ team
+%% @copyright 2023 VMware, Inc. or its affiliates.
+%%
+%% @doc
+%% This module provides an API to manage deprecated features in RabbitMQ. It
+%% is built on top of the Feature flags subsystem.
+%%
+%% == What a deprecated feature is ==
+%%
+%% A <strong>deprecated feature</strong> is a name and several properties
+%% given to a feature in RabbitMQ that will be removed in a future version. By
+%% defining a deprecated feature, we can communicate to end users that what
+%% they are using is going away and allow them to test how RabbitMQ behaves as
+%% if the feature was already removed.
+%%
+%% Because it is based on feature flags, everything in {@link
+%% rabbit_feature_flags} applies. However, the semantic is kind of reversed:
+%% when the feature flag behind a deprecated feature is enabled, this means the
+%% deprecated feature is removed or can be considered removed. Therefore
+%% <strong>the use of a deprecated feature is permitted while the backing
+%% feature flag is disabled and denied once the feature flag is
+%% enabled</strong>.
+%%
+%% == How to declare a deprecated feature ==
+%%
+%% To define a deprecated feature, you need to use the
+%% `-rabbit_deprecated_feature()' module attribute:
+%%
+%% ```
+%% -rabbit_deprecated_feature(DeprecatedFeature).
+%% '''
+%%
+%% `DeprecatedFeature' is a {@type deprecated_feature_modattr()}.
+%%
+%% == How to check that a deprecated feature is permitted ==
+%%
+%% To check in the code if a deprecated feature is permitted:
+%%
+%% ```
+%% case rabbit_deprecated_features:is_permitted(DeprecatedFeatureName) of
+%%     true ->
+%%         %% The deprecated feature is still permitted.
+%%         ok;
+%%     false ->
+%%         %% The deprecated feature is gone or should be considered
+%%         %% unavailable.
+%%         error
+%% end.
+%% '''
+%%
+%% == How to permit or not a deprecated feature from the configuration ==
+%%
+%% The following configuration snippet permits one deprecated feature and
+%% denies another one:
+%%
+%% ```
+%% deprecated_features.permit.my_deprecated_feature_1 = true
+%% deprecated_features.permit.my_deprecated_feature_2 = false
+%% '''
+%%
+%% == Differences with regular feature flags ==
+%%
+%% Despite the fact that a deprecated feature is implemented as a feature flag
+%% behind the scene, there is a slight difference of behavior in the way a
+%% deprecated feature's feature flag is enabled.
+%%
+%% A regular feature flag is disabled during RabbitMQ startup, except if it is
+%% required or has been enabled in a previous run. If this is the first time
+%% RabbitMQ starts, all stable feature flags are enabled.
+%%
+%% A deprecated feature's feature flag is enabled or disabled on startup
+%% depending on its deprecation phase. During `permitted_by_default', it is
+%% disabled out-of-the-box, except if configured otherwise. During
+%% `denied_by_default', it is enabled out-of-the-box, except if configured
+%% otherwise. When `disconnected' or `removed', it is always enabled like a
+%% required regular feature flag. This logic is in the registry initialization
+%% code in {@link rabbit_ff_registry_factory:maybe_initialize_registry/3}.
+%%
+%% Later during the boot process, after plugins are loaded and the Feature
+%% flags subsystem refreshes known feature flags, we execute the {@link
+%% is_feature_used_callback()} callback. If the feature is used and the
+%% underlying feature flag was enabled, then the refresh fails and RabbitMQ
+%% fails to start.
+%%
+%% This callback is also used when the underlying feature flag is enabled
+%% later at runtime by calling {@link rabbit_feature_flags:enable/1} or during
+%% a cluster-wide sync of the feature flags states. Again, the underlying
+%% feature flag won't be enabled if the feature is used.
+%%
+%% Note that this callback is only used when in `permitted_by_default' or
+%% `denied_by_default': remember that `disconnected' and `removed' are the
+%% same as a required feature flag.
+%%
+%% Another difference is that the state of a deprecated feature's feature flag
+%% is not recorded in the {@link
+%% rabbit_feature_flags:enabled_feature_flags_list_file/0}. As said earlier,
+%% the state is always computed when the registry is initialized.
+
+-module(rabbit_deprecated_features).
+
+-include_lib("kernel/include/logger.hrl").
+-include_lib("stdlib/include/assert.hrl").
+
+-include_lib("rabbit_common/include/logging.hrl").
+
+-include("src/rabbit_feature_flags.hrl").
+
+-export([is_permitted/1,
+         get_phase/1,
+         get_warning/1]).
+-export([extend_properties/2,
+         should_be_permitted/2,
+         enable_underlying_feature_flag_cb/1]).
+
+-type deprecated_feature_modattr() :: {rabbit_feature_flags:feature_name(),
+                                       feature_props()}.
+%% The value of a `-rabbitmq_deprecated_feature()' module attribute used to
+%% declare a deprecated feature.
+%%
+%% Example:
+%% ```
+%% -rabbit_deprecated_feature(
+%%    {my_deprecated_feature_1,
+%%     #{deprecation_phase => permitted_by_default,
+%%       msg_when_permitted => "Feature 1 will be removed from RabbitMQ X.0"
+%%      }}).
+%% '''
+
+-type deprecation_phase() :: permitted_by_default |
+                             denied_by_default |
+                             disconnected |
+                             removed.
+%% The deprecation phase of a feature.
+%%
+%% Deprecation phases are used in the following order:
+%% <ol>
+%% <li>`permitted_by_default': the feature is enabled by default and the user
+%% can use it like they did so far. They can turn it off in the configuration
+%% to experiment with the absence of the feature.</li>
+%% <li>`denied_by_default': the feature is disabled by default and the user
+%% must enable it in the configuration to be able to use it.</li>
+%% <li>`disconnected': the code of the feature is still there but it is
+%% disabled and can't be re-enabled from the configuration. The user has to
+%% recompile RabbitMQ to re-enable it.</li>
+%% <li>`removed': the code of the feature is no longer in the product. There is
+%% no way to re-enable it at this point. The deprecated feature must still be
+%% defined because, like required feature flags, its presence is important to
+%% determine if nodes are compatible and can be clustered together.</li>
+%% </ol>
+
+-type feature_props() :: #{desc => string(),
+                           doc_url => string(),
+                           deprecation_phase := deprecation_phase(),
+                           messages => #{when_permitted => string(),
+                                         when_denied => string(),
+                                         when_removed => string()},
+                           callbacks =>
+                           #{callback_name() =>
+                             rabbit_feature_flags:callback_fun_name()}}.
+%% The deprecated feature properties.
+%%
+%% The properties are:
+%% <ul>
+%% <li>`deprecation_phase': where the deprecated feature is in its
+%% lifecycle</li>
+%% <li>`messages': a map of warning/error messages for each situation:
+%% <ul>
+%% <li>`when_permitted': what message to log and possibly display to the user
+%% when the feature is being permitted and used. It is logged as a warning or
+%% displayed to the user by the CLI or in the management UI for instance.</li>
+%% <li>`when_denied': like `when_permitted', message used when an attempt to
+%% use a denied deprecated feature is being made. It is logged as an error or
+%% displayed to the user by the CLI or in the management UI for instance.</li>
+%% </ul></li>
+%% </ul>
+%%
+%% Other properties are the same as {@link
+%% rabbit_feature_flags:feature_props()}.
+
+-type feature_props_extended() ::
+      #{name := rabbit_feature_flags:feature_name(),
+        desc => string(),
+        doc_url => string(),
+        callbacks => #{callback_name() | enable =>
+                       rabbit_feature_flags:callback_fun_name()},
+        deprecation_phase := deprecation_phase(),
+        messages := #{when_permitted => string(),
+                      when_denied => string(),
+                      when_removed => string()},
+        provided_by := atom()}.
+%% The deprecated feature properties, once expanded by this module when
+%% feature flags are discovered.
+%%
+%% We make sure messages are set, possibly generating them automatically if
+%% needed. Other added properties are the same as {@link
+%% rabbit_feature_flags:feature_props_extended()}.
+
+-type callbacks() :: is_feature_used_callback().
+%% All possible callbacks.
+
+-type callbacks_args() :: is_feature_used_callback_args().
+%% All possible callbacks arguments.
+
+-type callbacks_rets() :: is_feature_used_callback_ret().
+%% All possible callbacks return values.
+
+-type callback_name() :: is_feature_used.
+%% Name of the callback.
+
+-type is_feature_used_callback() :: fun((is_feature_used_callback_args())
+                                        -> is_feature_used_callback_ret()).
+%% The callback called when a deprecated feature is about to be denied.
+%%
+%% If this callback returns true (i.e. the feature is currently actively
+%% used), the deprecated feature won't be marked as denied. This will also
+%% prevent the RabbitMQ node from starting.
+
+-type is_feature_used_callback_args() ::
+      #{feature_name := rabbit_feature_flags:feature_name(),
+        feature_props := feature_props_extended(),
+        command := is_feature_used,
+        nodes := [node()]}.
+%% A map passed to {@type is_feature_used_callback()}.
+
+-type is_feature_used_callback_ret() :: boolean().
+%% Return value of the `is_feature_used' callback.
+
+-export_type([deprecated_feature_modattr/0,
+              deprecation_phase/0,
+              feature_props/0,
+              feature_props_extended/0,
+              callbacks/0,
+              callback_name/0,
+              callbacks_args/0,
+              callbacks_rets/0,
+              is_feature_used_callback/0,
+              is_feature_used_callback_args/0,
+              is_feature_used_callback_ret/0]).
+
+%% -------------------------------------------------------------------
+%% Public API.
+%% -------------------------------------------------------------------
+
+-spec is_permitted(FeatureName) -> IsPermitted when
+      FeatureName :: rabbit_feature_flags:feature_name(),
+      IsPermitted :: boolean().
+%% @doc Indicates if the given deprecated feature is permitted or not.
+%%
+%% Calling this function automatically logs a warning or an error to let the
+%% user know they are using something that is or will be removed. For a given
+%% deprecated feature, automatic warning is limited to one occurence per day.
+%%
+%% @param FeatureName the name of the deprecated feature.
+%%
+%% @returns true if the deprecated feature can be used, false otherwise.
+
+is_permitted(FeatureName) ->
+    Permitted = is_permitted_nolog(FeatureName),
+    maybe_log_warning(FeatureName, Permitted),
+    Permitted.
+
+is_permitted_nolog(FeatureName) ->
+    not rabbit_feature_flags:is_enabled(FeatureName).
+
+-spec get_phase
+(FeatureName) -> Phase | undefined when
+      FeatureName :: rabbit_feature_flags:feature_name(),
+      Phase :: deprecation_phase();
+(FeatureProps) -> Phase when
+      FeatureProps :: feature_props() | feature_props_extended(),
+      Phase :: deprecation_phase().
+%% @doc Returns the deprecation phase of the given deprecated feature.
+%%
+%% @param FeatureName the name of the deprecated feature.
+%% @param FeatureProps the properties of the deprecated feature.
+%%
+%% @returns the deprecation phase, or `undefined' if the deprecated feature
+%% was given by its name and this name corresponds to no known deprecated
+%% features.
+
+get_phase(FeatureName) when is_atom(FeatureName) ->
+    case rabbit_ff_registry:get(FeatureName) of
+        undefined    -> undefined;
+        FeatureProps -> get_phase(FeatureProps)
+    end;
+get_phase(FeatureProps) when is_map(FeatureProps) ->
+    ?assert(?IS_DEPRECATION(FeatureProps)),
+    maps:get(deprecation_phase, FeatureProps).
+
+-spec get_warning
+(FeatureName) -> Warning | undefined when
+      FeatureName :: rabbit_feature_flags:feature_name(),
+      Warning :: string() | undefined;
+(FeatureProps) -> Warning when
+      FeatureProps :: feature_props_extended(),
+      Warning :: string().
+%% @doc Returns the message associated with the given deprecated feature.
+%%
+%% Messages are set in the `msg_when_permitted' and `msg_when_denied'
+%% properties.
+%%
+%% If the deprecated feature defines no warning in its declaration, a warning
+%% message is generated automatically.
+%%
+%% @param FeatureName the name of the deprecated feature.
+%% @param FeatureProps the properties of the deprecated feature.
+%%
+%% @returns the warning message, or `undefined' if the deprecated feature was
+%% given by its name and this name corresponds to no known deprecated
+%% features.
+
+get_warning(FeatureName) when is_atom(FeatureName) ->
+    case rabbit_ff_registry:get(FeatureName) of
+        undefined    -> undefined;
+        FeatureProps -> get_warning(FeatureProps)
+    end;
+get_warning(FeatureProps) when is_map(FeatureProps) ->
+    ?assert(?IS_DEPRECATION(FeatureProps)),
+    #{name := FeatureName} = FeatureProps,
+    Permitted = is_permitted_nolog(FeatureName),
+    get_warning(FeatureProps, Permitted).
+
+get_warning(FeatureName, Permitted) when is_atom(FeatureName) ->
+    case rabbit_ff_registry:get(FeatureName) of
+        undefined    -> undefined;
+        FeatureProps -> get_warning(FeatureProps, Permitted)
+    end;
+get_warning(FeatureProps, Permitted) when is_map(FeatureProps) ->
+    ?assert(?IS_DEPRECATION(FeatureProps)),
+    Phase = get_phase(FeatureProps),
+    Msgs = maps:get(messages, FeatureProps),
+    if
+        Phase =:= permitted_by_default orelse Phase =:= denied_by_default ->
+            case Permitted of
+                true  -> maps:get(when_permitted, Msgs);
+                false -> maps:get(when_denied, Msgs)
+            end;
+        true ->
+            maps:get(when_removed, Msgs)
+    end.
+
+%% -------------------------------------------------------------------
+%% Internal functions.
+%% -------------------------------------------------------------------
+
+-spec extend_properties(FeatureName, FeatureProps) -> ExtFeatureProps when
+      FeatureName :: rabbit_feature_flags:feature_name(),
+      FeatureProps :: feature_props() |
+                      feature_props_extended(),
+      ExtFeatureProps :: feature_props_extended().
+%% @doc Extend the deprecated feature properties.
+%%
+%% <ol>
+%% <li>It generates warning/error messages automatically if the properties
+%% don't have them set.</li>
+%% <li>It wraps the `is_feature_used' callback.</li>
+%% </ol>
+%%
+%% @private
+
+extend_properties(FeatureName, FeatureProps)
+  when ?IS_DEPRECATION(FeatureProps) ->
+    FeatureProps1 = generate_warnings(FeatureName, FeatureProps),
+    FeatureProps2 = wrap_callback(FeatureName, FeatureProps1),
+    FeatureProps2.
+
+generate_warnings(FeatureName, FeatureProps) ->
+    Msgs0 = maps:get(messages, FeatureProps, #{}),
+    Msgs1 = generate_warnings1(FeatureName, FeatureProps, Msgs0),
+    FeatureProps#{messages => Msgs1}.
+
+generate_warnings1(FeatureName, FeatureProps, Msgs) ->
+    Phase = get_phase(FeatureProps),
+    DefaultMsgs =
+    if
+        Phase =:= permitted_by_default ->
+            #{when_permitted =>
+              rabbit_misc:format(
+                "Feature `~ts` is deprecated.~n"
+                "By default, this feature can still be used for now.~n"
+                "Its use will not be permitted by default in a future minor"
+                "RabbitMQ version and the feature will be removed from a"
+                "future major RabbitMQ version; actual versions to be"
+                "determined.~n"
+                "To continue using this feature when it is not permitted "
+                "by default, set the following parameter in your "
+                "configuration:~n"
+                "    \"deprecated_features.permit.~ts = true\"~n"
+                "To test RabbitMQ as if the feature was removed, set this "
+                "in your configuration:~n"
+                "    \"deprecated_features.permit.~ts = false\"",
+                [FeatureName, FeatureName, FeatureName]),
+
+              when_denied =>
+              rabbit_misc:format(
+                "Feature `~ts` is deprecated.~n"
+                "Its use is not permitted per the configuration "
+                "(overriding the default, which is permitted):~n"
+                "    \"deprecated_features.permit.~ts = false\"~n"
+                "Its use will not be permitted by default in a future minor "
+                "RabbitMQ version and the feature will be removed from a "
+                "future major RabbitMQ version; actual versions to be "
+                "determined.~n"
+                "To continue using this feature when it is not permitted "
+                "by default, set the following parameter in your "
+                "configuration:~n"
+                "    \"deprecated_features.permit.~ts = true\"",
+                [FeatureName, FeatureName, FeatureName])};
+
+        Phase =:= denied_by_default ->
+            #{when_permitted =>
+              rabbit_misc:format(
+                "Feature `~ts` is deprecated.~n"
+                "Its use is permitted per the configuration (overriding "
+                "the default, which is not permitted):~n"
+                "    \"deprecated_features.permit.~ts = true\"~n"
+                "The feature will be removed from a future major RabbitMQ "
+                "version, regardless of the configuration.",
+                [FeatureName, FeatureName]),
+
+              when_denied =>
+              rabbit_misc:format(
+                "Feature `~ts` is deprecated.~n"
+                "By default, this feature is not permitted anymore.~n"
+                "The feature will be removed from a future major RabbitMQ "
+                "version, regardless of the configuration; actual version "
+                "to be determined.~n"
+                "To continue using this feature when it is not permitted "
+                "by default, set the following parameter in your "
+                "configuration:~n"
+                "    \"deprecated_features.permit.~ts = true\"",
+                [FeatureName, FeatureName])};
+
+        Phase =:= disconnected orelse Phase =:= removed ->
+            #{when_removed =>
+              rabbit_misc:format(
+                "Feature `~ts` is removed; "
+                "its use is not possible anymore.~n"
+                "If RabbitMQ refuses to start because of this, you need to "
+                "downgrade RabbitMQ and make sure the feature is not used "
+                "at all before upgrading again.",
+                [FeatureName])}
+    end,
+    maps:merge(DefaultMsgs, Msgs).
+
+wrap_callback(_FeatureName, #{callbacks := Callbacks} = FeatureProps) ->
+    Callbacks1 = Callbacks#{
+                   enable => {?MODULE, enable_underlying_feature_flag_cb}},
+    FeatureProps#{callbacks => Callbacks1};
+wrap_callback(_FeatureName, FeatureProps) ->
+    FeatureProps.
+
+-spec should_be_permitted(FeatureName, FeatureProps) -> IsPermitted when
+      FeatureName :: rabbit_feature_flags:feature_name(),
+      FeatureProps :: feature_props_extended(),
+      IsPermitted :: boolean().
+%% @doc Indicates if the deprecated feature should be permitted.
+%%
+%% The decision is based on the deprecation phase and the configuration.
+%%
+%% @private
+
+should_be_permitted(FeatureName, FeatureProps) ->
+    case get_phase(FeatureProps) of
+        permitted_by_default ->
+            is_permitted_in_configuration(FeatureName, true);
+        denied_by_default ->
+            is_permitted_in_configuration(FeatureName, false);
+        Phase ->
+            case is_permitted_in_configuration(FeatureName, false) of
+                true ->
+                    ?LOG_WARNING(
+                       "Deprecated features: `~ts`: ~ts feature, it "
+                       "cannot be permitted from configuration",
+                       [FeatureName, Phase],
+                       #{domain => ?RMQLOG_DOMAIN_FEAT_FLAGS});
+                false ->
+                    ok
+            end,
+            false
+    end.
+
+-spec is_permitted_in_configuration(FeatureName, Default) -> IsPermitted when
+      FeatureName :: rabbit_feature_flags:feature_name(),
+      Default :: boolean(),
+      IsPermitted :: boolean().
+%% @private
+
+is_permitted_in_configuration(FeatureName, Default) ->
+    Settings = application:get_env(rabbit, permit_deprecated_features, #{}),
+    case maps:get(FeatureName, Settings, undefined) of
+        undefined ->
+            Default;
+        Default ->
+            PermittedStr = case Default of
+                               true  -> "permitted";
+                               false -> "not permitted"
+                           end,
+            ?LOG_DEBUG(
+               "Deprecated features: `~ts`: ~ts in configuration, same as "
+               "default",
+               [FeatureName, PermittedStr],
+               #{domain => ?RMQLOG_DOMAIN_FEAT_FLAGS}),
+            Default;
+        Permitted ->
+            PermittedStr = case Permitted of
+                               true  -> "permitted";
+                               false -> "not permitted"
+                           end,
+            ?LOG_DEBUG(
+               "Deprecated features: `~ts`: ~ts in configuration, overrides "
+               "default",
+               [FeatureName, PermittedStr],
+               #{domain => ?RMQLOG_DOMAIN_FEAT_FLAGS}),
+            ?assert(is_boolean(Permitted)),
+            Permitted
+    end.
+
+-spec maybe_log_warning(FeatureName, Permitted) -> ok when
+      FeatureName :: rabbit_feature_flags:feature_name(),
+      Permitted :: boolean().
+%% @private
+
+maybe_log_warning(FeatureName, Permitted) ->
+    case should_log_warning(FeatureName) of
+        false ->
+            ok;
+        true ->
+            Warning = get_warning(FeatureName, Permitted),
+            FormatStr = "Deprecated features: `~ts`: ~ts",
+            FormatArgs = [FeatureName, Warning],
+            case Permitted of
+                true ->
+                    ?LOG_WARNING(
+                       FormatStr, FormatArgs,
+                       #{domain => ?RMQLOG_DOMAIN_FEAT_FLAGS});
+                false ->
+                    ?LOG_ERROR(
+                       FormatStr, FormatArgs,
+                       #{domain => ?RMQLOG_DOMAIN_FEAT_FLAGS})
+            end
+    end.
+
+-define(PT_DEPRECATION_WARNING_TS(FeatureName), {?MODULE, FeatureName}).
+
+-spec should_log_warning(FeatureName) -> ShouldLog when
+      FeatureName :: rabbit_feature_flags:feature_name(),
+      ShouldLog :: boolean().
+%% @private
+
+should_log_warning(FeatureName) ->
+    Key = ?PT_DEPRECATION_WARNING_TS(FeatureName),
+    Now = erlang:timestamp(),
+    try
+        Last = persistent_term:get(Key),
+        Diff = timer:now_diff(Now, Last),
+        if
+            Diff >= 24 * 60 * 60 * 1000 * 1000 ->
+                persistent_term:put(Key, Now),
+                true;
+            true ->
+                false
+        end
+    catch
+        error:badarg ->
+            persistent_term:put(Key, Now),
+            true
+    end.
+
+enable_underlying_feature_flag_cb(
+  #{command := enable,
+    feature_name := FeatureName,
+    feature_props := #{callbacks := Callbacks}} = Args) ->
+    case Callbacks of
+        #{is_feature_used := {CallbackMod, CallbackFun}} ->
+            Args1 = Args#{command => is_feature_used},
+            IsUsed = erlang:apply(CallbackMod, CallbackFun, [Args1]),
+            case IsUsed of
+                false ->
+                    ok;
+                true ->
+                    ?LOG_ERROR(
+                       "Deprecated features: `~ts`: can't deny deprecated "
+                       "feature because it is actively used",
+                       [FeatureName],
+                       #{domain => ?RMQLOG_DOMAIN_FEAT_FLAGS}),
+                    {error,
+                     {failed_to_deny_deprecated_features, [FeatureName]}}
+            end;
+        _ ->
+            ok
+    end.
diff --git a/deps/rabbit/src/rabbit_feature_flags.erl b/deps/rabbit/src/rabbit_feature_flags.erl
index 3f35fc47a9..9bf5a59c8c 100644
--- a/deps/rabbit/src/rabbit_feature_flags.erl
+++ b/deps/rabbit/src/rabbit_feature_flags.erl
@@ -82,6 +82,8 @@
 
 -include_lib("rabbit_common/include/logging.hrl").
 
+-include("src/rabbit_feature_flags.hrl").
+
 -export([list/0,
          list/1,
          list/2,
@@ -126,11 +128,7 @@
          get_overriden_running_nodes/0]).
 -endif.
 
-%% Default timeout for operations on remote nodes.
--define(TIMEOUT, 60000).
-
--type feature_flag_modattr() :: {feature_name(),
-                                 feature_props()}.
+-type feature_flag_modattr() :: {feature_name(), feature_props()}.
 %% The value of a `-rabbitmq_feature_flag()' module attribute used to
 %% declare a new feature flag.
 
@@ -158,7 +156,7 @@
 %% <li>`stability': the level of stability</li>
 %% <li>`depends_on': a list of feature flags name which must be enabled
 %%   before this one</li>
-%% <li>`callbacks': a map of callback names</li>
+%% <li>`callbacks': a map of callbacks</li>
 %% </ul>
 %%
 %% Note that each `callbacks' is a {@type callback_fun_name()}, not a {@type
@@ -167,12 +165,16 @@
 %% represent it as an Erlang term when we regenerate the registry module
 %% source code (using {@link erl_syntax:abstract/1}).
 
--type feature_flags() :: #{feature_name() => feature_props_extended()}.
+-type feature_flags() ::
+      #{feature_name() =>
+        feature_props_extended() |
+        rabbit_deprecated_features:feature_props_extended()}.
 %% The feature flags map as returned or accepted by several functions in
 %% this module. In particular, this what the {@link list/0} function
 %% returns.
 
--type feature_props_extended() :: #{desc => string(),
+-type feature_props_extended() :: #{name := feature_name(),
+                                    desc => string(),
                                     doc_url => string(),
                                     stability => stability(),
                                     depends_on => [feature_name()],
@@ -336,11 +338,23 @@ list() -> list(all).
 %% `disabled'.
 %% @returns A map of selected feature flags.
 
-list(all)      -> rabbit_ff_registry:list(all);
-list(enabled)  -> rabbit_ff_registry:list(enabled);
-list(disabled) -> maps:filter(
-                    fun(FeatureName, _) -> is_disabled(FeatureName) end,
-                    list(all)).
+list(all) ->
+    maps:filter(
+      fun(_, FeatureProps) -> ?IS_FEATURE_FLAG(FeatureProps) end,
+      rabbit_ff_registry:list(all));
+list(enabled) ->
+    maps:filter(
+      fun(_, FeatureProps) -> ?IS_FEATURE_FLAG(FeatureProps) end,
+      rabbit_ff_registry:list(enabled));
+list(disabled) ->
+    maps:filter(
+      fun
+          (FeatureName, FeatureProps) when ?IS_FEATURE_FLAG(FeatureProps) ->
+              is_disabled(FeatureName);
+          (_, _) ->
+              false
+      end,
+      list(all)).
 
 -spec list(all | enabled | disabled, stability()) -> feature_flags().
 %% @doc
@@ -354,13 +368,19 @@ list(disabled) -> maps:filter(
 %% @returns A map of selected feature flags.
 
 list(Which, stable) ->
-    maps:filter(fun(_, FeatureProps) ->
+    maps:filter(fun
+                    (_, FeatureProps) when ?IS_FEATURE_FLAG(FeatureProps) ->
                         Stability = get_stability(FeatureProps),
-                        stable =:= Stability orelse required =:= Stability
+                        stable =:= Stability orelse required =:= Stability;
+                    (_, _) ->
+                        false
                 end, list(Which));
 list(Which, experimental) ->
-    maps:filter(fun(_, FeatureProps) ->
-                        experimental =:= get_stability(FeatureProps)
+    maps:filter(fun
+                    (_, FeatureProps) when ?IS_FEATURE_FLAG(FeatureProps) ->
+                        experimental =:= get_stability(FeatureProps);
+                    (_, _) ->
+                        false
                 end, list(Which)).
 
 -spec enable(feature_name() | [feature_name()]) -> ok |
@@ -687,7 +707,9 @@ get_state(FeatureName) when is_atom(FeatureName) ->
       FeatureName :: feature_name(),
       Stability :: stability();
 (FeatureProps) -> Stability when
-      FeatureProps :: feature_props_extended(),
+      FeatureProps ::
+      feature_props_extended() |
+      rabbit_deprecated_features:feature_props_extended(),
       Stability :: stability().
 %% @doc
 %% Returns the stability of a feature flag.
@@ -714,8 +736,16 @@ get_stability(FeatureName) when is_atom(FeatureName) ->
         undefined    -> undefined;
         FeatureProps -> get_stability(FeatureProps)
     end;
-get_stability(FeatureProps) when is_map(FeatureProps) ->
-    maps:get(stability, FeatureProps, stable).
+get_stability(FeatureProps) when  ?IS_FEATURE_FLAG(FeatureProps) ->
+    maps:get(stability, FeatureProps, stable);
+get_stability(FeatureProps) when ?IS_DEPRECATION(FeatureProps) ->
+    Phase = rabbit_deprecated_features:get_phase(FeatureProps),
+    case Phase of
+        removed              -> required;
+        disconnected         -> required;
+        denied_by_default    -> stable;
+        permitted_by_default -> experimental
+    end.
 
 %% -------------------------------------------------------------------
 %% Feature flags registry.
@@ -751,9 +781,12 @@ inject_test_feature_flags(FeatureFlags) ->
                                                  _ ->
                                                      '$injected'
                                              end,
+                                    FeatureProps1 = maps:remove(
+                                                      provided_by,
+                                                      FeatureProps),
                                     FFlags0 = maps:get(Origin, Acc, #{}),
                                     FFlags1 = FFlags0#{
-                                                FeatureName => FeatureProps},
+                                                FeatureName => FeatureProps1},
                                     Acc#{Origin => FFlags1}
                             end, FeatureFlagsPerApp0, FeatureFlags),
     AttributesFromTestsuite = maps:fold(
@@ -788,23 +821,38 @@ query_supported_feature_flags() ->
     %% application might be loaded/present and not have a specific feature
     %% flag. In this case, the feature flag should be considered unsupported.
     ScannedApps = rabbit_misc:rabbitmq_related_apps(),
-    AttributesPerApp = rabbit_misc:module_attributes_from_apps(
-                         rabbit_feature_flag, ScannedApps),
-    AttributesFromTestsuite = module_attributes_from_testsuite(),
-    TestsuiteProviders = [App || {App, _, _} <- AttributesFromTestsuite],
+    AttrsPerAppA = rabbit_misc:module_attributes_from_apps(
+                    rabbit_feature_flag, ScannedApps),
+    AttrsPerAppB = rabbit_misc:module_attributes_from_apps(
+                    rabbit_deprecated_feature, ScannedApps),
+    AttrsFromTestsuite = module_attributes_from_testsuite(),
+    TestsuiteProviders = [App || {App, _, _} <- AttrsFromTestsuite],
     T1 = erlang:timestamp(),
     ?LOG_DEBUG(
-      "Feature flags: time to find supported feature flags: ~tp us",
+      "Feature flags: time to find supported feature flags and deprecated "
+      "features: ~tp us",
       [timer:now_diff(T1, T0)],
       #{domain => ?RMQLOG_DOMAIN_FEAT_FLAGS}),
-    AllAttributes = AttributesPerApp ++ AttributesFromTestsuite,
+    AllAttributes = AttrsPerAppA ++ AttrsPerAppB ++ AttrsFromTestsuite,
     AllApps = lists:usort(ScannedApps ++ TestsuiteProviders),
     {AllApps, prepare_queried_feature_flags(AllAttributes, #{})}.
 
+-spec prepare_queried_feature_flags(AllAttributes, AllFeatureFlags) ->
+    AllFeatureFlags when
+      AllAttributes :: [{App, Module, Attributes}],
+      App :: atom(),
+      Module :: module(),
+      Attributes ::
+      [feature_flag_modattr() |
+       rabbit_deprecated_features:deprecated_feature_modattr()],
+      AllFeatureFlags :: feature_flags().
+%% @private
+
 prepare_queried_feature_flags([{App, _Module, Attributes} | Rest],
                               AllFeatureFlags) ->
     ?LOG_DEBUG(
-      "Feature flags: application `~ts` has ~b feature flags",
+      "Feature flags: application `~ts` has ~b feature flags (including "
+      "deprecated features)",
       [App, length(Attributes)],
       #{domain => ?RMQLOG_DOMAIN_FEAT_FLAGS}),
     AllFeatureFlags1 = lists:foldl(
@@ -820,32 +868,76 @@ prepare_queried_feature_flags([{App, _Module, Attributes} | Rest],
 prepare_queried_feature_flags([], AllFeatureFlags) ->
     AllFeatureFlags.
 
+-spec assert_feature_flag_is_valid(FeatureName, FeatureProps) -> ok when
+      FeatureName :: feature_name(),
+      FeatureProps :: feature_props() |
+                      rabbit_deprecated_features:feature_props().
+%% @private
+
 assert_feature_flag_is_valid(FeatureName, FeatureProps) ->
     try
         ?assert(is_atom(FeatureName)),
         ?assert(is_map(FeatureProps)),
-        Stability = get_stability(FeatureProps),
-        ?assert(Stability =:= stable orelse
-                Stability =:= experimental orelse
-                Stability =:= required),
-        ?assertNot(maps:is_key(migration_fun, FeatureProps)),
-        case FeatureProps of
-            #{callbacks := Callbacks} ->
-                Known = [enable,
-                         post_enable],
-                ?assert(is_map(Callbacks)),
-                ?assertEqual([], maps:keys(Callbacks) -- Known),
-                lists:foreach(
-                  fun(CallbackMF) ->
-                          ?assertMatch({_, _}, CallbackMF),
-                          {CallbackMod, CallbackFun} = CallbackMF,
-                          ?assert(is_atom(CallbackMod)),
-                          ?assert(is_atom(CallbackFun)),
-                          ?assert(erlang:function_exported(
-                                    CallbackMod, CallbackFun, 1))
-                  end, maps:values(Callbacks));
-            _ ->
-                ok
+        ?assert(is_list(maps:get(desc, FeatureProps, ""))),
+        ?assert(is_list(maps:get(doc_url, FeatureProps, ""))),
+        if
+            ?IS_FEATURE_FLAG(FeatureProps) ->
+                ValidProps = [desc,
+                              doc_url,
+                              stability,
+                              depends_on,
+                              callbacks],
+                ?assertEqual([], maps:keys(FeatureProps) -- ValidProps),
+                ?assert(is_list(maps:get(depends_on, FeatureProps, []))),
+                ?assert(lists:all(
+                          fun erlang:is_atom/1,
+                          maps:get(depends_on, FeatureProps, []))),
+                Stability = get_stability(FeatureProps),
+                ?assert(Stability =:= stable orelse
+                        Stability =:= experimental orelse
+                        Stability =:= required),
+                ?assertNot(maps:is_key(migration_fun, FeatureProps)),
+                ?assertNot(maps:is_key(warning, FeatureProps)),
+                case FeatureProps of
+                    #{callbacks := Callbacks} ->
+                        ValidCbs = [enable,
+                                    post_enable],
+                        ?assert(is_map(Callbacks)),
+                        ?assertEqual([], maps:keys(Callbacks) -- ValidCbs),
+                        assert_callbacks_are_valid(Callbacks);
+                    _ ->
+                        ok
+                end;
+            ?IS_DEPRECATION(FeatureProps) ->
+                ValidProps = [desc,
+                              doc_url,
+                              deprecation_phase,
+                              messages,
+                              callbacks],
+                ?assertEqual([], maps:keys(FeatureProps) -- ValidProps),
+                Phase = maps:get(deprecation_phase, FeatureProps),
+                ?assert(Phase =:= permitted_by_default orelse
+                        Phase =:= denied_by_default orelse
+                        Phase =:= disconnected orelse
+                        Phase =:= removed),
+                Msgs = maps:get(messages, FeatureProps, #{}),
+                ?assert(is_map(Msgs)),
+                ValidMsgs = [when_permitted,
+                             when_denied,
+                             when_removed],
+                ?assertEqual([], maps:keys(Msgs) -- ValidMsgs),
+                ?assert(lists:all(fun io_lib:char_list/1, maps:values(Msgs))),
+                ?assertNot(maps:is_key(stability, FeatureProps)),
+                ?assertNot(maps:is_key(migration_fun, FeatureProps)),
+                case FeatureProps of
+                    #{callbacks := Callbacks} ->
+                        ValidCbs = [is_feature_used],
+                        ?assert(is_map(Callbacks)),
+                        ?assertEqual([], maps:keys(Callbacks) -- ValidCbs),
+                        assert_callbacks_are_valid(Callbacks);
+                    _ ->
+                        ok
+                end
         end
     catch
         Class:Reason:Stacktrace ->
@@ -860,21 +952,51 @@ assert_feature_flag_is_valid(FeatureName, FeatureProps) ->
             erlang:raise(Class, Reason, Stacktrace)
     end.
 
--spec merge_new_feature_flags(feature_flags(),
-                              atom(),
-                              feature_name(),
-                              feature_props()) -> feature_flags().
+assert_callbacks_are_valid(Callbacks) ->
+    lists:foreach(
+      fun(CallbackMF) ->
+              ?assertMatch({_, _}, CallbackMF),
+              {CallbackMod, CallbackFun} = CallbackMF,
+              ?assert(is_atom(CallbackMod)),
+              ?assert(is_atom(CallbackFun)),
+              %% Make sure the module is loaded before we check the function
+              %% is exported.
+              _ = CallbackMod:module_info(),
+              ?assert(erlang:function_exported(
+                        CallbackMod, CallbackFun, 1))
+      end, maps:values(Callbacks)).
+
+-spec merge_new_feature_flags(FeatureFlags,
+                              App,
+                              FeatureName,
+                              FeatureProps) -> FeatureFlags when
+      FeatureFlags :: feature_flags(),
+      App :: atom(),
+      FeatureName :: feature_name(),
+      FeatureProps :: feature_props() |
+                      rabbit_deprecated_features:feature_props(),
+      FeatureFlags :: feature_flags().
 %% @private
 
 merge_new_feature_flags(AllFeatureFlags, App, FeatureName, FeatureProps)
   when is_atom(FeatureName) andalso is_map(FeatureProps) ->
-    %% We expand the feature flag properties map with:
+    %% We extend the feature flag properties map with:
+    %%   - the name of the feature flag itself, just in case some code has
+    %%     access to the feature props only.
     %%   - the name of the application providing it: only informational
     %%     for now, but can be handy to understand that a feature flag
     %%     comes from a plugin.
-    FeatureProps1 = maps:put(provided_by, App, FeatureProps),
+    FeatureProps1 = FeatureProps#{name => FeatureName,
+                                  provided_by => App},
+    FeatureProps2 = if
+                        ?IS_DEPRECATION(FeatureProps) ->
+                            rabbit_deprecated_features:extend_properties(
+                              FeatureName, FeatureProps1);
+                        true ->
+                            FeatureProps1
+                    end,
     maps:merge(AllFeatureFlags,
-               #{FeatureName => FeatureProps1}).
+               #{FeatureName => FeatureProps2}).
 
 %% -------------------------------------------------------------------
 %% Feature flags state storage.
@@ -970,17 +1092,26 @@ try_to_write_enabled_feature_flags_list(FeatureNames) ->
                             {error, _} -> [];
                             List       -> List
                         end,
-    FeatureNames1 = lists:foldl(
+    FeatureNames1 = lists:filter(
+                      fun(FeatureName) ->
+                              case rabbit_ff_registry:get(FeatureName) of
+                                  undefined ->
+                                      false;
+                                  FeatureProps ->
+                                      ?IS_FEATURE_FLAG(FeatureProps)
+                              end
+                      end, FeatureNames),
+    FeatureNames2 = lists:foldl(
                       fun(Name, Acc) ->
                               case is_supported_locally(Name) of
                                   true  -> Acc;
                                   false -> [Name | Acc]
                               end
-                      end, FeatureNames, PreviouslyEnabled),
-    FeatureNames2 = lists:sort(FeatureNames1),
+                      end, FeatureNames1, PreviouslyEnabled),
+    FeatureNames3 = lists:sort(FeatureNames2),
 
     File = enabled_feature_flags_list_file(),
-    Content = io_lib:format("~tp.~n", [FeatureNames2]),
+    Content = io_lib:format("~tp.~n", [FeatureNames3]),
     %% TODO: If we fail to write the the file, we should spawn a process
     %% to retry the operation.
     case file:write_file(File, Content) of
diff --git a/deps/rabbit/src/rabbit_feature_flags.hrl b/deps/rabbit/src/rabbit_feature_flags.hrl
new file mode 100644
index 0000000000..071ec4ff65
--- /dev/null
+++ b/deps/rabbit/src/rabbit_feature_flags.hrl
@@ -0,0 +1,7 @@
+-define(
+   IS_FEATURE_FLAG(FeatureProps),
+   (is_map(FeatureProps) andalso not ?IS_DEPRECATION(FeatureProps))).
+
+-define(
+   IS_DEPRECATION(FeatureProps),
+   (is_map(FeatureProps) andalso is_map_key(deprecation_phase, FeatureProps))).
diff --git a/deps/rabbit/src/rabbit_ff_controller.erl b/deps/rabbit/src/rabbit_ff_controller.erl
index cc46ad5d24..55ee02f165 100644
--- a/deps/rabbit/src/rabbit_ff_controller.erl
+++ b/deps/rabbit/src/rabbit_ff_controller.erl
@@ -31,6 +31,8 @@
 
 -include_lib("rabbit_common/include/logging.hrl").
 
+-include("src/rabbit_feature_flags.hrl").
+
 -export([is_supported/1, is_supported/2,
          enable/1,
          enable_default/0,
@@ -61,6 +63,8 @@
 -record(?MODULE, {from,
                   notify = #{}}).
 
+-type run_callback_error() :: {error, any()}.
+
 -define(LOCAL_NAME, ?MODULE).
 -define(GLOBAL_NAME, {?MODULE, global}).
 
@@ -488,11 +492,13 @@ enable_default_task() ->
             #{feature_flags := FeatureFlags} = Inventory,
             StableFeatureNames =
             maps:fold(
-              fun
-                  (FeatureName, #{stability := stable}, Acc) ->
-                      [FeatureName | Acc];
-                  (_FeatureName, _FeatureProps, Acc) ->
-                      Acc
+              fun(FeatureName, FeatureProps, Acc) ->
+                      Stability = rabbit_feature_flags:get_stability(
+                                    FeatureProps),
+                      case Stability of
+                          stable -> [FeatureName | Acc];
+                          _      -> Acc
+                      end
               end, [], FeatureFlags),
             enable_many(Inventory, StableFeatureNames);
         [] ->
@@ -617,9 +623,32 @@ sync_cluster_task() ->
 
     case collect_inventory_on_nodes(Nodes) of
         {ok, Inventory} ->
-            FeatureNames = list_feature_flags_enabled_somewhere(
-                             Inventory, false),
-            enable_many(Inventory, FeatureNames);
+            CantEnable = list_deprecated_features_that_cant_be_denied(
+                           Inventory),
+            case CantEnable of
+                [] ->
+                    FeatureNames = list_feature_flags_enabled_somewhere(
+                                     Inventory, false),
+                    enable_many(Inventory, FeatureNames);
+                _ ->
+                    ?LOG_ERROR(
+                       "Feature flags: the following deprecated features "
+                       "can't be denied because their associated feature "
+                       "is being actively used: ~0tp",
+                       [CantEnable],
+                       #{domain => ?RMQLOG_DOMAIN_FEAT_FLAGS}),
+                    lists:foreach(
+                      fun(FeatureName) ->
+                              Warning =
+                              rabbit_deprecated_features:get_warning(
+                                FeatureName),
+                              ?LOG_ERROR(
+                                 "~ts", [Warning],
+                                 #{domain => ?RMQLOG_DOMAIN_FEAT_FLAGS})
+                      end, CantEnable),
+                    {error,
+                     {failed_to_deny_deprecated_features, CantEnable}}
+            end;
         Error ->
             Error
     end.
@@ -1001,7 +1030,8 @@ collect_inventory_on_nodes(Nodes, Timeout) ->
 merge_feature_flags(FeatureFlagsA, FeatureFlagsB) ->
     FeatureFlags = maps:merge(FeatureFlagsA, FeatureFlagsB),
     maps:map(
-      fun(FeatureName, FeatureProps) ->
+      fun
+          (FeatureName, FeatureProps) when ?IS_FEATURE_FLAG(FeatureProps) ->
               %% When we collect feature flag properties from all nodes, we
               %% start with an empty cluster inventory (a common Erlang
               %% recursion pattern). This means that all feature flags are
@@ -1040,7 +1070,28 @@ merge_feature_flags(FeatureFlagsA, FeatureFlagsB) ->
 
               FeatureProps1 = FeatureProps#{stability => Stability},
               FeatureProps2 = maps:remove(callbacks, FeatureProps1),
-              FeatureProps2
+              FeatureProps2;
+          (FeatureName, FeatureProps) when ?IS_DEPRECATION(FeatureProps) ->
+              UnknownProps = #{deprecation_phase => permitted_by_default},
+              FeaturePropsA = maps:get(
+                                FeatureName, FeatureFlagsA, UnknownProps),
+              FeaturePropsB = maps:get(
+                                FeatureName, FeatureFlagsB, UnknownProps),
+
+              PhaseA = rabbit_deprecated_features:get_phase(FeaturePropsA),
+              PhaseB = rabbit_deprecated_features:get_phase(FeaturePropsB),
+              Phase = case {PhaseA, PhaseB} of
+                          {removed, _}           -> removed;
+                          {_, removed}           -> removed;
+                          {disconnected, _}      -> disconnected;
+                          {_, disconnected}      -> disconnected;
+                          {denied_by_default, _} -> denied_by_default;
+                          {_, denied_by_default} -> denied_by_default;
+                          _                      -> permitted_by_default
+                      end,
+
+              FeatureProps1 = FeatureProps#{deprecation_phase => Phase},
+              FeatureProps1
       end, FeatureFlags).
 
 -spec list_feature_flags_enabled_somewhere(Inventory, HandleStateChanging) ->
@@ -1070,6 +1121,32 @@ list_feature_flags_enabled_somewhere(
                      end, #{}, StatesPerNode),
     lists:sort(maps:keys(MergedStates)).
 
+-spec list_deprecated_features_that_cant_be_denied(Inventory) ->
+    Ret when
+      Inventory :: rabbit_feature_flags:cluster_inventory(),
+      Ret :: [FeatureName],
+      FeatureName :: rabbit_feature_flags:feature_name().
+
+list_deprecated_features_that_cant_be_denied(
+  #{states_per_node := StatesPerNode}) ->
+    ThisNode = node(),
+    States = maps:get(ThisNode, StatesPerNode),
+
+    maps:fold(
+      fun
+          (FeatureName, true, Acc) ->
+              #{ThisNode := IsUsed} = run_callback(
+                                        [ThisNode], FeatureName,
+                                        is_feature_used, #{}, infinity),
+              case IsUsed of
+                  true   -> [FeatureName | Acc];
+                  false  -> Acc;
+                  _Error -> Acc
+              end;
+          (_FeatureName, false, Acc) ->
+              Acc
+      end, [], States).
+
 -spec list_nodes_who_know_the_feature_flag(Inventory, FeatureName) ->
     Ret when
       Inventory :: rabbit_feature_flags:cluster_inventory(),
@@ -1188,14 +1265,17 @@ rpc_calls(Nodes, Module, Function, Args, Timeout) when is_list(Nodes) ->
 %% Feature flag support queries.
 %% --------------------------------------------------------------------
 
--spec is_known(Inventory, FeatureFlag) -> IsKnown when
+-spec is_known(Inventory, FeatureProps) -> IsKnown when
       Inventory :: rabbit_feature_flags:cluster_inventory(),
-      FeatureFlag :: rabbit_feature_flags:feature_props_extended(),
+      FeatureProps ::
+      rabbit_feature_flags:feature_props_extended() |
+      rabbit_deprecated_features:feature_props_extended(),
       IsKnown :: boolean().
+%% @private
 
 is_known(
   #{applications_per_node := ScannedAppsPerNode},
-  #{provided_by := App} = _FeatureFlag) ->
+  #{provided_by := App} = _FeatureProps) ->
     maps:fold(
       fun
           (_Node, ScannedApps, false) -> lists:member(App, ScannedApps);
@@ -1329,14 +1409,34 @@ enable_dependencies1(
 %% Migration function.
 %% --------------------------------------------------------------------
 
--spec run_callback(Nodes, FeatureName, Command, Extra, Timeout) ->
-    Rets when
+-spec run_callback
+(Nodes, FeatureName, Command, Extra, Timeout) -> Rets when
       Nodes :: [node()],
       FeatureName :: rabbit_feature_flags:feature_name(),
-      Command :: rabbit_feature_flags:callback_name(),
+      Command :: enable,
       Extra :: map(),
       Timeout :: timeout(),
-      Rets :: #{node() => term()}.
+      Rets :: #{node() =>
+                rabbit_feature_flags:enable_callback_ret() |
+                run_callback_error()};
+(Nodes, FeatureName, Command, Extra, Timeout) -> Rets when
+      Nodes :: [node()],
+      FeatureName :: rabbit_feature_flags:feature_name(),
+      Command :: post_enable,
+      Extra :: map(),
+      Timeout :: timeout(),
+      Rets :: #{node() =>
+                rabbit_feature_flags:post_enable_callback_ret() |
+                run_callback_error()};
+(Nodes, FeatureName, Command, Extra, Timeout) -> Rets when
+      Nodes :: [node()],
+      FeatureName :: rabbit_feature_flags:feature_name(),
+      Command :: is_feature_used,
+      Extra :: map(),
+      Timeout :: timeout(),
+      Rets :: #{node() =>
+                rabbit_deprecated_features:is_feature_used_callback_ret() |
+                run_callback_error()}.
 
 run_callback(Nodes, FeatureName, Command, Extra, Timeout) ->
     FeatureProps = rabbit_ff_registry:get(FeatureName),
@@ -1364,8 +1464,9 @@ run_callback(Nodes, FeatureName, Command, Extra, Timeout) ->
             %% No callbacks defined for this feature flag. Consider it a
             %% success!
             Ret = case Command of
-                      enable      -> ok;
-                      post_enable -> ok
+                      enable          -> ok;
+                      post_enable     -> ok;
+                      is_feature_used -> false
                   end,
             #{node() => Ret}
     end.
@@ -1375,9 +1476,12 @@ run_callback(Nodes, FeatureName, Command, Extra, Timeout) ->
       Nodes :: [node()],
       CallbackMod :: module(),
       CallbackFun :: atom(),
-      Args :: rabbit_feature_flags:callbacks_args(),
+      Args :: rabbit_feature_flags:callbacks_args() |
+              rabbit_deprecated_features:callbacks_args(),
       Timeout :: timeout(),
-      Rets :: #{node() => rabbit_feature_flags:callbacks_rets()}.
+      Rets :: #{node() => rabbit_feature_flags:callbacks_rets() |
+                          rabbit_deprecated_features:callbacks_rets() |
+                          run_callback_error()}.
 
 do_run_callback(Nodes, CallbackMod, CallbackFun, Args, Timeout) ->
     #{feature_name := FeatureName,
diff --git a/deps/rabbit/src/rabbit_ff_registry.erl b/deps/rabbit/src/rabbit_ff_registry.erl
index 26718e2cc7..f62e828f85 100644
--- a/deps/rabbit/src/rabbit_ff_registry.erl
+++ b/deps/rabbit/src/rabbit_ff_registry.erl
@@ -62,8 +62,12 @@
                 end
         end).
 
--spec get(rabbit_feature_flags:feature_name()) ->
-    rabbit_feature_flags:feature_props_extended() | undefined.
+-spec get
+(FeatureName) -> FeatureProps | undefined when
+      FeatureName :: rabbit_feature_flags:feature_name(),
+      FeatureProps ::
+      rabbit_feature_flags:feature_props_extended() |
+      rabbit_deprecated_features:feature_props_extended().
 %% @doc
 %% Returns the properties of a feature flag.
 %%
@@ -78,9 +82,24 @@ get(FeatureName) ->
     ?convince_dialyzer(
        ?MODULE:get(FeatureName),
        undefined,
-       #{provided_by => rabbit}).
-
--spec list(all | enabled | disabled) -> rabbit_feature_flags:feature_flags().
+       lists:nth(
+         rand:uniform(2),
+         [#{name => feature_flag,
+            provided_by => rabbit},
+          #{name => deprecated_feature,
+            deprecation_phase =>
+            lists:nth(
+              4,
+              [permitted_by_default,
+               denied_by_default,
+               disconnected,
+               removed]),
+            messages => #{},
+            provided_by => rabbit}])).
+
+-spec list(Which) -> FeatureFlags when
+      Which :: all | enabled | disabled,
+      FeatureFlags :: rabbit_feature_flags:feature_flags().
 %% @doc
 %% Lists all, enabled or disabled feature flags, depending on the argument.
 %%
@@ -95,7 +114,8 @@ list(Which) ->
     _ = rabbit_ff_registry_factory:initialize_registry(),
     ?convince_dialyzer(?MODULE:list(Which), #{}, #{}).
 
--spec states() -> rabbit_feature_flags:feature_states().
+-spec states() -> FeatureStates when
+      FeatureStates :: rabbit_feature_flags:feature_states().
 %% @doc
 %% Returns the states of supported feature flags.
 %%
@@ -108,7 +128,9 @@ states() ->
     _ = rabbit_ff_registry_factory:initialize_registry(),
     ?convince_dialyzer(?MODULE:states(), #{}, #{}).
 
--spec is_supported(rabbit_feature_flags:feature_name()) -> boolean().
+-spec is_supported(FeatureName) -> IsSupported when
+      FeatureName :: rabbit_feature_flags:feature_name(),
+      IsSupported :: boolean().
 %% @doc
 %% Returns if a feature flag is supported.
 %%
@@ -123,7 +145,9 @@ is_supported(FeatureName) ->
     _ = rabbit_ff_registry_factory:initialize_registry(),
     ?convince_dialyzer(?MODULE:is_supported(FeatureName), false, true).
 
--spec is_enabled(rabbit_feature_flags:feature_name()) -> boolean() | state_changing.
+-spec is_enabled(FeatureName) -> IsEnabled | state_changing when
+      FeatureName :: rabbit_feature_flags:feature_name(),
+      IsEnabled :: boolean().
 %% @doc
 %% Returns if a feature flag is enabled or if its state is changing.
 %%
@@ -138,7 +162,8 @@ is_enabled(FeatureName) ->
     _ = rabbit_ff_registry_factory:initialize_registry(),
     ?convince_dialyzer(?MODULE:is_enabled(FeatureName), false, true).
 
--spec is_registry_initialized() -> boolean().
+-spec is_registry_initialized() -> IsInitialized when
+      IsInitialized :: boolean().
 %% @doc
 %% Indicates if the registry is initialized.
 %%
@@ -152,7 +177,8 @@ is_enabled(FeatureName) ->
 is_registry_initialized() ->
     always_return_false().
 
--spec is_registry_written_to_disk() -> boolean().
+-spec is_registry_written_to_disk() -> WrittenToDisk when
+      WrittenToDisk :: boolean().
 %% @doc
 %% Indicates if the feature flags state was successfully persisted to disk.
 %%
@@ -169,7 +195,8 @@ is_registry_initialized() ->
 is_registry_written_to_disk() ->
     always_return_true().
 
--spec inventory() -> rabbit_feature_flags:inventory().
+-spec inventory() -> Inventory when
+      Inventory :: rabbit_feature_flags:inventory().
 
 inventory() ->
     _ = rabbit_ff_registry_factory:initialize_registry(),
diff --git a/deps/rabbit/src/rabbit_ff_registry_factory.erl b/deps/rabbit/src/rabbit_ff_registry_factory.erl
index 83c6e9b2ff..3a7c956560 100644
--- a/deps/rabbit/src/rabbit_ff_registry_factory.erl
+++ b/deps/rabbit/src/rabbit_ff_registry_factory.erl
@@ -12,6 +12,8 @@
 
 -include_lib("rabbit_common/include/logging.hrl").
 
+-include("src/rabbit_feature_flags.hrl").
+
 -export([initialize_registry/0,
          initialize_registry/1,
          initialize_registry/3,
@@ -27,28 +29,32 @@
 
 -type registry_vsn() :: term().
 
--spec acquire_state_change_lock() -> boolean().
+-spec acquire_state_change_lock() -> ok.
+
 acquire_state_change_lock() ->
     ?LOG_DEBUG(
       "Feature flags: acquiring lock ~tp",
       [?FF_STATE_CHANGE_LOCK],
       #{domain => ?RMQLOG_DOMAIN_FEAT_FLAGS}),
-    Ret = global:set_lock(?FF_STATE_CHANGE_LOCK),
+    true = global:set_lock(?FF_STATE_CHANGE_LOCK),
     ?LOG_DEBUG(
       "Feature flags: acquired lock ~tp",
       [?FF_STATE_CHANGE_LOCK],
       #{domain => ?RMQLOG_DOMAIN_FEAT_FLAGS}),
-    Ret.
+    ok.
+
+-spec release_state_change_lock() -> ok.
 
--spec release_state_change_lock() -> true.
 release_state_change_lock() ->
     ?LOG_DEBUG(
       "Feature flags: releasing lock ~tp",
       [?FF_STATE_CHANGE_LOCK],
       #{domain => ?RMQLOG_DOMAIN_FEAT_FLAGS}),
-    global:del_lock(?FF_STATE_CHANGE_LOCK).
+    true = global:del_lock(?FF_STATE_CHANGE_LOCK),
+    ok.
 
--spec initialize_registry() -> ok | {error, any()} | no_return().
+-spec initialize_registry() -> Ret when
+      Ret :: ok | {error, any()} | no_return().
 %% @private
 %% @doc
 %% Initializes or reinitializes the registry.
@@ -69,8 +75,9 @@ release_state_change_lock() ->
 initialize_registry() ->
     initialize_registry(#{}).
 
--spec initialize_registry(rabbit_feature_flags:feature_flags()) ->
-    ok | {error, any()} | no_return().
+-spec initialize_registry(FeatureFlags) -> Ret when
+      FeatureFlags :: rabbit_feature_flags:feature_flags(),
+      Ret :: ok | {error, any()} | no_return().
 %% @private
 %% @doc
 %% Initializes or reinitializes the registry.
@@ -135,10 +142,13 @@ initialize_registry(NewSupportedFeatureFlags) ->
 enabled_feature_flags_to_feature_states(FeatureNames) ->
     maps:from_list([{FeatureName, true} || FeatureName <- FeatureNames]).
 
--spec initialize_registry(rabbit_feature_flags:feature_flags(),
-                          rabbit_feature_flags:feature_states(),
-                          boolean()) ->
-    ok | {error, any()} | no_return().
+-spec initialize_registry(FeatureFlags,
+                          FeatureStates,
+                          WrittenToDisk) -> Ret when
+      FeatureFlags :: rabbit_feature_flags:feature_flags(),
+      FeatureStates :: rabbit_feature_flags:feature_states(),
+      WrittenToDisk :: boolean(),
+      Ret :: ok | {error, any()} | no_return().
 %% @private
 %% @doc
 %% Initializes or reinitializes the registry.
@@ -173,10 +183,13 @@ initialize_registry(NewSupportedFeatureFlags,
             Error2
     end.
 
--spec maybe_initialize_registry(rabbit_feature_flags:feature_flags(),
-                                rabbit_feature_flags:feature_states(),
-                                boolean()) ->
-    ok | restart | {error, any()} | no_return().
+-spec maybe_initialize_registry(FeatureFlags,
+                                FeatureStates,
+                                WrittenToDisk) -> Ret when
+      FeatureFlags :: rabbit_feature_flags:feature_flags(),
+      FeatureStates :: rabbit_feature_flags:feature_states(),
+      WrittenToDisk :: boolean(),
+      Ret :: ok | restart | {error, any()} | no_return().
 
 maybe_initialize_registry(NewSupportedFeatureFlags,
                           NewFeatureStates,
@@ -243,53 +256,55 @@ maybe_initialize_registry(NewSupportedFeatureFlags,
                          false ->
                              NewFeatureStates
                      end,
-    FeatureStates = maps:map(
-                      fun(FeatureName, FeatureProps) ->
-                              Stability = maps:get(
-                                            stability, FeatureProps, stable),
-                              ProvidedBy = maps:get(
-                                             provided_by, FeatureProps),
-                              State = case FeatureStates0 of
-                                          #{FeatureName := FeatureState} ->
-                                              FeatureState;
-                                          _ ->
-                                              false
-                                      end,
-                              case Stability of
-                                  required when State =:= true ->
-                                      %% The required feature flag is already
-                                      %% enabled, we keep it this way.
-                                      State;
-                                  required when NewNode ->
-                                      %% This is the very first time the node
-                                      %% starts, we already mark the required
-                                      %% feature flag as enabled.
-                                      ?assertNotEqual(state_changing, State),
-                                      true;
-                                  required when ProvidedBy =/= rabbit ->
-                                      ?assertNotEqual(state_changing, State),
-                                      true;
-                                  required ->
-                                      %% This is not a new node and the
-                                      %% required feature flag is disabled.
-                                      %% This is an error and RabbitMQ must be
-                                      %% downgraded to enable the feature
-                                      %% flag.
-                                      ?assertNotEqual(state_changing, State),
-                                      ?LOG_ERROR(
-                                        "Feature flags: `~ts`: required "
-                                        "feature flag not enabled! It must "
-                                        "be enabled before upgrading "
-                                        "RabbitMQ.",
-                                        [FeatureName],
-                                        #{domain => ?RMQLOG_DOMAIN_FEAT_FLAGS}),
-                                      throw({error,
-                                             {disabled_required_feature_flag,
-                                              FeatureName}});
-                                  _ ->
-                                      State
-                              end
-                      end, AllFeatureFlags),
+    FeatureStates =
+    maps:map(
+      fun
+          (FeatureName, FeatureProps) when ?IS_FEATURE_FLAG(FeatureProps) ->
+              Stability = rabbit_feature_flags:get_stability(FeatureProps),
+              ProvidedBy = maps:get(provided_by, FeatureProps),
+              State = case FeatureStates0 of
+                          #{FeatureName := FeatureState} -> FeatureState;
+                          _                              -> false
+                      end,
+              case Stability of
+                  required when State =:= true ->
+                      %% The required feature flag is already enabled, we keep
+                      %% it this way.
+                      State;
+                  required when NewNode ->
+                      %% This is the very first time the node starts, we
+                      %% already mark the required feature flag as enabled.
+                      ?assertNotEqual(state_changing, State),
+                      true;
+                  required when ProvidedBy =/= rabbit ->
+                      ?assertNotEqual(state_changing, State),
+                      true;
+                  required ->
+                      %% This is not a new node and the required feature flag
+                      %% is disabled. This is an error and RabbitMQ must be
+                      %% downgraded to enable the feature flag.
+                      ?assertNotEqual(state_changing, State),
+                      ?LOG_ERROR(
+                         "Feature flags: `~ts`: required feature flag not "
+                         "enabled! It must be enabled before upgrading "
+                         "RabbitMQ.",
+                         [FeatureName],
+                         #{domain => ?RMQLOG_DOMAIN_FEAT_FLAGS}),
+                      throw({error,
+                             {disabled_required_feature_flag,
+                              FeatureName}});
+                  _ ->
+                      State
+              end;
+          (FeatureName, FeatureProps) when ?IS_DEPRECATION(FeatureProps) ->
+              case FeatureStates0 of
+                  #{FeatureName := FeatureState} ->
+                      FeatureState;
+                  _ ->
+                      not rabbit_deprecated_features:should_be_permitted(
+                            FeatureName, FeatureProps)
+              end
+      end, AllFeatureFlags),
 
     %% The feature flags inventory is used by rabbit_ff_controller to query
     %% feature flags atomically. The inventory also contains the list of
@@ -329,10 +344,13 @@ maybe_initialize_registry(NewSupportedFeatureFlags,
             ok
     end.
 
--spec does_registry_need_refresh(rabbit_feature_flags:feature_flags(),
-                                 rabbit_feature_flags:feature_states(),
-                                 boolean()) ->
-    boolean().
+-spec does_registry_need_refresh(FeatureFlags,
+                                 FeatureStates,
+                                 WrittenToDisk) -> Ret when
+      FeatureFlags :: rabbit_feature_flags:feature_flags(),
+      FeatureStates :: rabbit_feature_flags:feature_states(),
+      WrittenToDisk :: boolean(),
+      Ret :: boolean().
 
 does_registry_need_refresh(AllFeatureFlags,
                            FeatureStates,
@@ -380,12 +398,17 @@ does_registry_need_refresh(AllFeatureFlags,
             true
     end.
 
--spec do_initialize_registry(registry_vsn(),
-                             rabbit_feature_flags:feature_flags(),
-                             rabbit_feature_flags:feature_states(),
-                             rabbit_feature_flags:inventory(),
-                             boolean()) ->
-    ok | restart | {error, any()} | no_return().
+-spec do_initialize_registry(Vsn,
+                             FeatureFlags,
+                             FeatureStates,
+                             Inventory,
+                             WrittenToDisk) -> Ret when
+      Vsn :: registry_vsn(),
+      FeatureFlags :: rabbit_feature_flags:feature_flags(),
+      FeatureStates :: rabbit_feature_flags:feature_states(),
+      Inventory :: rabbit_feature_flags:inventory(),
+      WrittenToDisk :: boolean(),
+      Ret :: ok | restart | {error, any()} | no_return().
 %% @private
 
 do_initialize_registry(RegistryVsn,
@@ -395,8 +418,19 @@ do_initialize_registry(RegistryVsn,
                        WrittenToDisk) ->
     %% We log the state of those feature flags.
     ?LOG_DEBUG(
-      "Feature flags: list of feature flags found:\n" ++
       lists:flatten(
+        "Feature flags: list of feature flags found:\n" ++
+        [io_lib:format(
+           "Feature flags:   [~ts] ~ts~n",
+           [case maps:get(FeatureName, FeatureStates, false) of
+                true           -> "x";
+                state_changing -> "~";
+                false          -> " "
+            end,
+            FeatureName])
+         || FeatureName <- lists:sort(maps:keys(AllFeatureFlags)),
+            ?IS_FEATURE_FLAG(maps:get(FeatureName, AllFeatureFlags))] ++
+        "Feature flags: list of deprecated features found:\n" ++
         [io_lib:format(
            "Feature flags:   [~ts] ~ts~n",
            [case maps:get(FeatureName, FeatureStates, false) of
@@ -405,7 +439,8 @@ do_initialize_registry(RegistryVsn,
                 false          -> " "
             end,
             FeatureName])
-         || FeatureName <- lists:sort(maps:keys(AllFeatureFlags))] ++
+         || FeatureName <- lists:sort(maps:keys(AllFeatureFlags)),
+            ?IS_DEPRECATION(maps:get(FeatureName, AllFeatureFlags))] ++
         [io_lib:format(
            "Feature flags: scanned applications: ~tp~n"
            "Feature flags: feature flag states written to disk: ~ts",
@@ -663,8 +698,11 @@ maybe_log_registry_source_code(Forms) ->
 registry_loading_lock() -> ?FF_REGISTRY_LOADING_LOCK.
 -endif.
 
--spec load_registry_mod(registry_vsn(), module(), binary()) ->
-    ok | restart | no_return().
+-spec load_registry_mod(Vsn, Mod, Bin) -> Ret when
+      Vsn :: registry_vsn(),
+      Mod :: module(),
+      Bin :: binary(),
+      Ret :: ok | restart | no_return().
 %% @private
 
 load_registry_mod(RegistryVsn, Mod, Bin) ->
@@ -726,7 +764,8 @@ load_registry_mod(RegistryVsn, Mod, Bin) ->
             throw({feature_flag_registry_reload_failure, Reason})
     end.
 
--spec registry_vsn() -> registry_vsn().
+-spec registry_vsn() -> Vsn when
+      Vsn :: registry_vsn().
 %% @private
 
 registry_vsn() ->
diff --git a/deps/rabbit/test/config_schema_SUITE_data/rabbit.snippets b/deps/rabbit/test/config_schema_SUITE_data/rabbit.snippets
index c6c1cc917f..76c1c63a1e 100644
--- a/deps/rabbit/test/config_schema_SUITE_data/rabbit.snippets
+++ b/deps/rabbit/test/config_schema_SUITE_data/rabbit.snippets
@@ -905,7 +905,7 @@ credential_validator.regexp = ^abc\\d+",
   {deprecated_features_cmq,
    "deprecated_features.permit.classic_mirrored_queues = false",
    [{rabbit, [
-      {permitted_deprecated_features, #{classic_mirrored_queues => false}}
+      {permit_deprecated_features, #{classic_mirrored_queues => false}}
      ]}],
    []}
 
diff --git a/deps/rabbit/test/deprecated_features_SUITE.erl b/deps/rabbit/test/deprecated_features_SUITE.erl
new file mode 100644
index 0000000000..3dc4cf328f
--- /dev/null
+++ b/deps/rabbit/test/deprecated_features_SUITE.erl
@@ -0,0 +1,633 @@
+%% 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(deprecated_features_SUITE).
+
+-include_lib("eunit/include/eunit.hrl").
+-include_lib("common_test/include/ct.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,
+
+         use_unknown_deprecated_feature/1,
+         use_deprecated_feature_permitted_by_default_everywhere/1,
+         use_deprecated_feature_denied_by_default_everywhere/1,
+         use_deprecated_feature_disconnected_everywhere/1,
+         use_deprecated_feature_removed_everywhere/1,
+         override_permitted_by_default_in_configuration/1,
+         override_denied_by_default_in_configuration/1,
+         override_disconnected_in_configuration/1,
+         override_removed_in_configuration/1,
+         has_is_feature_used_cb_returning_false/1,
+         has_is_feature_used_cb_returning_true/1,
+         get_appropriate_warning_when_permitted/1,
+         get_appropriate_warning_when_denied/1,
+         get_appropriate_warning_when_disconnected/1,
+         get_appropriate_warning_when_removed/1,
+
+         feature_is_unused/1,
+         feature_is_used/1
+        ]).
+
+suite() ->
+    [{timetrap, {minutes, 1}}].
+
+all() ->
+    [
+     {group, cluster_size_1},
+     {group, cluster_size_3}
+    ].
+
+groups() ->
+    Tests = [
+             use_unknown_deprecated_feature,
+             use_deprecated_feature_permitted_by_default_everywhere,
+             use_deprecated_feature_denied_by_default_everywhere,
+             use_deprecated_feature_disconnected_everywhere,
+             use_deprecated_feature_removed_everywhere,
+             override_permitted_by_default_in_configuration,
+             override_denied_by_default_in_configuration,
+             override_disconnected_in_configuration,
+             override_removed_in_configuration,
+             has_is_feature_used_cb_returning_false,
+             has_is_feature_used_cb_returning_true,
+             get_appropriate_warning_when_permitted,
+             get_appropriate_warning_when_denied,
+             get_appropriate_warning_when_disconnected,
+             get_appropriate_warning_when_removed
+            ],
+    [
+     {cluster_size_1, [], Tests},
+     {cluster_size_3, [], Tests}
+    ].
+
+%% -------------------------------------------------------------------
+%% 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(cluster_size_1, Config) ->
+    rabbit_ct_helpers:set_config(Config, {nodes_count, 1});
+init_per_group(cluster_size_3, Config) ->
+    rabbit_ct_helpers:set_config(Config, {nodes_count, 3});
+init_per_group(_Group, Config) ->
+    Config.
+
+end_per_group(_Group, Config) ->
+    Config.
+
+init_per_testcase(Testcase, Config) ->
+    rabbit_ct_helpers:run_steps(
+      Config,
+      [fun(Cfg) ->
+               feature_flags_v2_SUITE:start_slave_nodes(Cfg, Testcase)
+       end]).
+
+end_per_testcase(_Testcase, Config) ->
+    rabbit_ct_helpers:run_steps(
+      Config,
+      [fun feature_flags_v2_SUITE:stop_slave_nodes/1]).
+
+%% -------------------------------------------------------------------
+%% Testcases.
+%% -------------------------------------------------------------------
+
+use_unknown_deprecated_feature(Config) ->
+    AllNodes = ?config(nodes, Config),
+    FeatureName = ?FUNCTION_NAME,
+    _ = [ok =
+         feature_flags_v2_SUITE:run_on_node(
+           Node,
+           fun() ->
+                   ?assertNot(rabbit_feature_flags:is_supported(FeatureName)),
+                   ?assertNot(rabbit_feature_flags:is_enabled(FeatureName)),
+                   ?assert(
+                      rabbit_deprecated_features:is_permitted(FeatureName)),
+
+                   %% The node doesn't know about the deprecated feature and
+                   %% thus rejects the request.
+                   ?assertEqual(
+                      {error, unsupported},
+                      rabbit_feature_flags:enable(FeatureName)),
+                   ?assertNot(rabbit_feature_flags:is_supported(FeatureName)),
+                   ?assertNot(rabbit_feature_flags:is_enabled(FeatureName)),
+                   ?assert(
+                      rabbit_deprecated_features:is_permitted(FeatureName)),
+                   ok
+           end)
+         || Node <- AllNodes].
+
+use_deprecated_feature_permitted_by_default_everywhere(Config) ->
+    [FirstNode | _] = AllNodes = ?config(nodes, Config),
+    feature_flags_v2_SUITE:connect_nodes(AllNodes),
+    feature_flags_v2_SUITE:override_running_nodes(AllNodes),
+
+    FeatureName = ?FUNCTION_NAME,
+    FeatureFlags = #{FeatureName =>
+                     #{provided_by => rabbit,
+                       deprecation_phase => permitted_by_default}},
+    ?assertEqual(
+       ok,
+       feature_flags_v2_SUITE:inject_on_nodes(AllNodes, FeatureFlags)),
+
+    _ = [ok =
+         feature_flags_v2_SUITE:run_on_node(
+           Node,
+           fun() ->
+                   ?assert(rabbit_feature_flags:is_supported(FeatureName)),
+                   ?assertNot(rabbit_feature_flags:is_enabled(FeatureName)),
+                   ?assert(
+                      rabbit_deprecated_features:is_permitted(FeatureName)),
+                   ok
+           end)
+         || Node <- AllNodes],
+
+    ok = feature_flags_v2_SUITE:run_on_node(
+           FirstNode,
+           fun() ->
+                   ?assertEqual(
+                      ok,
+                      rabbit_feature_flags:enable(FeatureName)),
+                   ok
+           end),
+
+    _ = [ok =
+         feature_flags_v2_SUITE:run_on_node(
+           Node,
+           fun() ->
+                   ?assert(rabbit_feature_flags:is_supported(FeatureName)),
+                   ?assert(rabbit_feature_flags:is_enabled(FeatureName)),
+                   ?assertNot(
+                      rabbit_deprecated_features:is_permitted(FeatureName)),
+                   ok
+           end )
+         || Node <- AllNodes].
+
+use_deprecated_feature_denied_by_default_everywhere(Config) ->
+    AllNodes = ?config(nodes, Config),
+    feature_flags_v2_SUITE:connect_nodes(AllNodes),
+    feature_flags_v2_SUITE:override_running_nodes(AllNodes),
+
+    FeatureName = ?FUNCTION_NAME,
+    FeatureFlags = #{FeatureName =>
+                     #{provided_by => rabbit,
+                       deprecation_phase => denied_by_default}},
+    ?assertEqual(
+       ok,
+       feature_flags_v2_SUITE:inject_on_nodes(AllNodes, FeatureFlags)),
+
+    _ = [ok =
+         feature_flags_v2_SUITE:run_on_node(
+           Node,
+           fun() ->
+                   ?assert(rabbit_feature_flags:is_supported(FeatureName)),
+                   ?assert(rabbit_feature_flags:is_enabled(FeatureName)),
+                   ?assertNot(
+                      rabbit_deprecated_features:is_permitted(FeatureName)),
+                   ok
+           end)
+         || Node <- AllNodes].
+
+use_deprecated_feature_disconnected_everywhere(Config) ->
+    AllNodes = ?config(nodes, Config),
+    feature_flags_v2_SUITE:connect_nodes(AllNodes),
+    feature_flags_v2_SUITE:override_running_nodes(AllNodes),
+
+    FeatureName = ?FUNCTION_NAME,
+    FeatureFlags = #{FeatureName =>
+                     #{provided_by => rabbit,
+                       deprecation_phase => disconnected}},
+    ?assertEqual(
+       ok,
+       feature_flags_v2_SUITE:inject_on_nodes(AllNodes, FeatureFlags)),
+
+    _ = [ok =
+         feature_flags_v2_SUITE:run_on_node(
+           Node,
+           fun() ->
+                   ?assert(rabbit_feature_flags:is_supported(FeatureName)),
+                   ?assert(rabbit_feature_flags:is_enabled(FeatureName)),
+                   ?assertNot(
+                      rabbit_deprecated_features:is_permitted(FeatureName)),
+                   ok
+           end)
+         || Node <- AllNodes].
+
+use_deprecated_feature_removed_everywhere(Config) ->
+    AllNodes = ?config(nodes, Config),
+    feature_flags_v2_SUITE:connect_nodes(AllNodes),
+    feature_flags_v2_SUITE:override_running_nodes(AllNodes),
+
+    FeatureName = ?FUNCTION_NAME,
+    FeatureFlags = #{FeatureName =>
+                     #{provided_by => rabbit,
+                       deprecation_phase => removed}},
+    ?assertEqual(
+       ok,
+       feature_flags_v2_SUITE:inject_on_nodes(AllNodes, FeatureFlags)),
+
+    _ = [ok =
+         feature_flags_v2_SUITE:run_on_node(
+           Node,
+           fun() ->
+                   ?assert(rabbit_feature_flags:is_supported(FeatureName)),
+                   ?assert(rabbit_feature_flags:is_enabled(FeatureName)),
+                   ?assertNot(
+                      rabbit_deprecated_features:is_permitted(FeatureName)),
+                   ok
+           end)
+         || Node <- AllNodes].
+
+override_permitted_by_default_in_configuration(Config) ->
+    AllNodes = ?config(nodes, Config),
+    feature_flags_v2_SUITE:connect_nodes(AllNodes),
+    feature_flags_v2_SUITE:override_running_nodes(AllNodes),
+
+    FeatureName = ?FUNCTION_NAME,
+    FeatureFlags = #{FeatureName =>
+                     #{provided_by => rabbit,
+                       deprecation_phase => permitted_by_default}},
+
+    _ = [ok =
+         feature_flags_v2_SUITE:run_on_node(
+           Node,
+           fun() ->
+                   application:set_env(
+                     rabbit, permit_deprecated_features,
+                     #{FeatureName => false}, [{persistent, false}])
+           end)
+         || Node <- AllNodes],
+
+    ?assertEqual(
+       ok,
+       feature_flags_v2_SUITE:inject_on_nodes(AllNodes, FeatureFlags)),
+
+    _ = [ok =
+         feature_flags_v2_SUITE:run_on_node(
+           Node,
+           fun() ->
+                   ?assert(rabbit_feature_flags:is_supported(FeatureName)),
+                   ?assert(rabbit_feature_flags:is_enabled(FeatureName)),
+                   ?assertNot(
+                      rabbit_deprecated_features:is_permitted(FeatureName)),
+                   ok
+           end)
+         || Node <- AllNodes].
+
+override_denied_by_default_in_configuration(Config) ->
+    AllNodes = ?config(nodes, Config),
+    feature_flags_v2_SUITE:connect_nodes(AllNodes),
+    feature_flags_v2_SUITE:override_running_nodes(AllNodes),
+
+    FeatureName = ?FUNCTION_NAME,
+    FeatureFlags = #{FeatureName =>
+                     #{provided_by => rabbit,
+                       deprecation_phase => denied_by_default}},
+
+    _ = [ok =
+         feature_flags_v2_SUITE:run_on_node(
+           Node,
+           fun() ->
+                   application:set_env(
+                     rabbit, permit_deprecated_features,
+                     #{FeatureName => true}, [{persistent, false}])
+           end)
+         || Node <- AllNodes],
+
+    ?assertEqual(
+       ok,
+       feature_flags_v2_SUITE:inject_on_nodes(AllNodes, FeatureFlags)),
+
+    _ = [ok =
+         feature_flags_v2_SUITE:run_on_node(
+           Node,
+           fun() ->
+                   ?assert(rabbit_feature_flags:is_supported(FeatureName)),
+                   ?assertNot(rabbit_feature_flags:is_enabled(FeatureName)),
+                   ?assert(
+                      rabbit_deprecated_features:is_permitted(FeatureName)),
+                   ok
+           end)
+         || Node <- AllNodes].
+
+override_disconnected_in_configuration(Config) ->
+    AllNodes = ?config(nodes, Config),
+    feature_flags_v2_SUITE:connect_nodes(AllNodes),
+    feature_flags_v2_SUITE:override_running_nodes(AllNodes),
+
+    FeatureName = ?FUNCTION_NAME,
+    FeatureFlags = #{FeatureName =>
+                     #{provided_by => rabbit,
+                       deprecation_phase => disconnected}},
+
+    _ = [ok =
+         feature_flags_v2_SUITE:run_on_node(
+           Node,
+           fun() ->
+                   application:set_env(
+                     rabbit, permit_deprecated_features,
+                     #{FeatureName => true}, [{persistent, false}])
+           end)
+         || Node <- AllNodes],
+
+    ?assertEqual(
+       ok,
+       feature_flags_v2_SUITE:inject_on_nodes(AllNodes, FeatureFlags)),
+
+    _ = [ok =
+         feature_flags_v2_SUITE:run_on_node(
+           Node,
+           fun() ->
+                   ?assert(rabbit_feature_flags:is_supported(FeatureName)),
+                   ?assert(rabbit_feature_flags:is_enabled(FeatureName)),
+                   ?assertNot(
+                      rabbit_deprecated_features:is_permitted(FeatureName)),
+                   ok
+           end)
+         || Node <- AllNodes].
+
+override_removed_in_configuration(Config) ->
+    AllNodes = ?config(nodes, Config),
+    feature_flags_v2_SUITE:connect_nodes(AllNodes),
+    feature_flags_v2_SUITE:override_running_nodes(AllNodes),
+
+    FeatureName = ?FUNCTION_NAME,
+    FeatureFlags = #{FeatureName =>
+                     #{provided_by => rabbit,
+                       deprecation_phase => removed}},
+
+    _ = [ok =
+         feature_flags_v2_SUITE:run_on_node(
+           Node,
+           fun() ->
+                   application:set_env(
+                     rabbit, permit_deprecated_features,
+                     #{FeatureName => true}, [{persistent, false}])
+           end)
+         || Node <- AllNodes],
+
+    ?assertEqual(
+       ok,
+       feature_flags_v2_SUITE:inject_on_nodes(AllNodes, FeatureFlags)),
+
+    _ = [ok =
+         feature_flags_v2_SUITE:run_on_node(
+           Node,
+           fun() ->
+                   ?assert(rabbit_feature_flags:is_supported(FeatureName)),
+                   ?assert(rabbit_feature_flags:is_enabled(FeatureName)),
+                   ?assertNot(
+                      rabbit_deprecated_features:is_permitted(FeatureName)),
+                   ok
+           end)
+         || Node <- AllNodes].
+
+has_is_feature_used_cb_returning_false(Config) ->
+    AllNodes = ?config(nodes, Config),
+    feature_flags_v2_SUITE:connect_nodes(AllNodes),
+    feature_flags_v2_SUITE:override_running_nodes(AllNodes),
+
+    FeatureName = ?FUNCTION_NAME,
+    FeatureFlags = #{FeatureName =>
+                     #{provided_by => rabbit,
+                       deprecation_phase => denied_by_default,
+                       callbacks => #{is_feature_used =>
+                                      {?MODULE, feature_is_unused}}}},
+    ?assertEqual(
+       ok,
+       feature_flags_v2_SUITE:inject_on_nodes(AllNodes, FeatureFlags)),
+
+    _ = [ok =
+         feature_flags_v2_SUITE:run_on_node(
+           Node,
+           fun() ->
+                   ?assert(rabbit_feature_flags:is_supported(FeatureName)),
+                   ?assert(rabbit_feature_flags:is_enabled(FeatureName)),
+                   ?assertNot(
+                      rabbit_deprecated_features:is_permitted(FeatureName)),
+                   ok
+           end)
+         || Node <- AllNodes].
+
+feature_is_unused(_Args) ->
+    false.
+
+has_is_feature_used_cb_returning_true(Config) ->
+    AllNodes = ?config(nodes, Config),
+    feature_flags_v2_SUITE:connect_nodes(AllNodes),
+    feature_flags_v2_SUITE:override_running_nodes(AllNodes),
+
+    FeatureName = ?FUNCTION_NAME,
+    FeatureFlags = #{FeatureName =>
+                     #{provided_by => rabbit,
+                       deprecation_phase => denied_by_default,
+                       callbacks => #{is_feature_used =>
+                                      {?MODULE, feature_is_used}}}},
+    ?assertEqual(
+       {error, {failed_to_deny_deprecated_features, [FeatureName]}},
+       feature_flags_v2_SUITE:inject_on_nodes(AllNodes, FeatureFlags)),
+
+    %% The deprecated feature is marked as denied when the registry is
+    %% initialized/updated. It is the refresh that will return an error (the
+    %% one returned above).
+    _ = [ok =
+         feature_flags_v2_SUITE:run_on_node(
+           Node,
+           fun() ->
+                   ?assert(rabbit_feature_flags:is_supported(FeatureName)),
+                   ?assert(rabbit_feature_flags:is_enabled(FeatureName)),
+                   ?assertNot(
+                      rabbit_deprecated_features:is_permitted(FeatureName)),
+                   ok
+           end)
+         || Node <- AllNodes].
+
+feature_is_used(_Args) ->
+    true.
+
+-define(MSGS, #{when_permitted => "permitted",
+                when_denied => "denied",
+                when_removed => "removed"}).
+
+get_appropriate_warning_when_permitted(Config) ->
+    [FirstNode | _] = AllNodes = ?config(nodes, Config),
+    feature_flags_v2_SUITE:connect_nodes(AllNodes),
+    feature_flags_v2_SUITE:override_running_nodes(AllNodes),
+
+    FeatureName = ?FUNCTION_NAME,
+    FeatureFlags = #{FeatureName =>
+                     #{provided_by => rabbit,
+                       deprecation_phase => permitted_by_default,
+                       messages => ?MSGS}},
+    ?assertEqual(
+       ok,
+       feature_flags_v2_SUITE:inject_on_nodes(AllNodes, FeatureFlags)),
+
+    _ = [ok =
+         feature_flags_v2_SUITE:run_on_node(
+           Node,
+           fun() ->
+                   ?assert(
+                      rabbit_deprecated_features:is_permitted(FeatureName)),
+                   ?assertEqual(
+                      maps:get(when_permitted, ?MSGS),
+                      rabbit_deprecated_features:get_warning(FeatureName)),
+                   ok
+           end)
+         || Node <- AllNodes],
+
+    ok = feature_flags_v2_SUITE:run_on_node(
+           FirstNode,
+           fun() ->
+                   ?assertEqual(
+                      ok,
+                      rabbit_feature_flags:enable(FeatureName)),
+                   ok
+           end),
+
+    _ = [ok =
+         feature_flags_v2_SUITE:run_on_node(
+           Node,
+           fun() ->
+                   ?assertNot(
+                      rabbit_deprecated_features:is_permitted(FeatureName)),
+                   ?assertEqual(
+                      maps:get(when_denied, ?MSGS),
+                      rabbit_deprecated_features:get_warning(FeatureName)),
+                   ok
+           end)
+         || Node <- AllNodes].
+
+get_appropriate_warning_when_denied(Config) ->
+    [FirstNode | _] = AllNodes = ?config(nodes, Config),
+    feature_flags_v2_SUITE:connect_nodes(AllNodes),
+    feature_flags_v2_SUITE:override_running_nodes(AllNodes),
+
+    FeatureName = ?FUNCTION_NAME,
+    FeatureFlags = #{FeatureName =>
+                     #{provided_by => rabbit,
+                       deprecation_phase => denied_by_default,
+                       messages => ?MSGS}},
+
+    _ = [ok =
+         feature_flags_v2_SUITE:run_on_node(
+           Node,
+           fun() ->
+                   application:set_env(
+                     rabbit, permit_deprecated_features,
+                     #{FeatureName => true}, [{persistent, false}])
+           end)
+         || Node <- AllNodes],
+
+    ?assertEqual(
+       ok,
+       feature_flags_v2_SUITE:inject_on_nodes(AllNodes, FeatureFlags)),
+
+    _ = [ok =
+         feature_flags_v2_SUITE:run_on_node(
+           Node,
+           fun() ->
+                   ?assert(
+                      rabbit_deprecated_features:is_permitted(FeatureName)),
+                   ?assertEqual(
+                      maps:get(when_permitted, ?MSGS),
+                      rabbit_deprecated_features:get_warning(FeatureName)),
+                   ok
+           end)
+         || Node <- AllNodes],
+
+    ok = feature_flags_v2_SUITE:run_on_node(
+           FirstNode,
+           fun() ->
+                   ?assertEqual(
+                      ok,
+                      rabbit_feature_flags:enable(FeatureName)),
+                   ok
+           end),
+
+    _ = [ok =
+         feature_flags_v2_SUITE:run_on_node(
+           Node,
+           fun() ->
+                   ?assertNot(
+                      rabbit_deprecated_features:is_permitted(FeatureName)),
+                   ?assertEqual(
+                      maps:get(when_denied, ?MSGS),
+                      rabbit_deprecated_features:get_warning(FeatureName)),
+                   ok
+           end)
+         || Node <- AllNodes].
+
+get_appropriate_warning_when_disconnected(Config) ->
+    AllNodes = ?config(nodes, Config),
+    feature_flags_v2_SUITE:connect_nodes(AllNodes),
+    feature_flags_v2_SUITE:override_running_nodes(AllNodes),
+
+    FeatureName = ?FUNCTION_NAME,
+    FeatureFlags = #{FeatureName =>
+                     #{provided_by => rabbit,
+                       deprecation_phase => disconnected,
+                       messages => ?MSGS}},
+    ?assertEqual(
+       ok,
+       feature_flags_v2_SUITE:inject_on_nodes(AllNodes, FeatureFlags)),
+
+    _ = [ok =
+         feature_flags_v2_SUITE:run_on_node(
+           Node,
+           fun() ->
+                   ?assertNot(
+                      rabbit_deprecated_features:is_permitted(FeatureName)),
+                   ?assertEqual(
+                      maps:get(when_removed, ?MSGS),
+                      rabbit_deprecated_features:get_warning(FeatureName)),
+                   ok
+           end)
+         || Node <- AllNodes].
+
+get_appropriate_warning_when_removed(Config) ->
+    AllNodes = ?config(nodes, Config),
+    feature_flags_v2_SUITE:connect_nodes(AllNodes),
+    feature_flags_v2_SUITE:override_running_nodes(AllNodes),
+
+    FeatureName = ?FUNCTION_NAME,
+    FeatureFlags = #{FeatureName =>
+                     #{provided_by => rabbit,
+                       deprecation_phase => disconnected,
+                       messages => ?MSGS}},
+    ?assertEqual(
+       ok,
+       feature_flags_v2_SUITE:inject_on_nodes(AllNodes, FeatureFlags)),
+
+    _ = [ok =
+         feature_flags_v2_SUITE:run_on_node(
+           Node,
+           fun() ->
+                   ?assertNot(
+                      rabbit_deprecated_features:is_permitted(FeatureName)),
+                   ?assertEqual(
+                      maps:get(when_removed, ?MSGS),
+                      rabbit_deprecated_features:get_warning(FeatureName)),
+                   ok
+           end)
+         || Node <- AllNodes].
diff --git a/deps/rabbit/test/feature_flags_v2_SUITE.erl b/deps/rabbit/test/feature_flags_v2_SUITE.erl
index dc1c6636b3..c16dd4f56e 100644
--- a/deps/rabbit/test/feature_flags_v2_SUITE.erl
+++ b/deps/rabbit/test/feature_flags_v2_SUITE.erl
@@ -23,6 +23,13 @@
          init_per_testcase/2,
          end_per_testcase/2,
 
+         start_slave_nodes/2,
+         stop_slave_nodes/1,
+         inject_on_nodes/2,
+         run_on_node/2,
+         connect_nodes/1,
+         override_running_nodes/1,
+
          mf_count_runs/1,
          mf_wait_and_count_runs_v2_enable/1,
          mf_wait_and_count_runs_v2_post_enable/1,
@@ -272,7 +279,12 @@ inject_on_nodes(Nodes, FeatureFlags) ->
            end,
            [])
          || Node <- Nodes],
-    ok.
+    run_on_node(
+      hd(Nodes),
+      fun() ->
+              rabbit_feature_flags:refresh_feature_flags_after_app_load()
+      end,
+      []).
 
 %% -------------------------------------------------------------------
 %% Migration functions.
@@ -419,7 +431,7 @@ enable_supported_feature_flag_in_a_3node_cluster(Config) ->
     FeatureName = ?FUNCTION_NAME,
     FeatureFlags = #{FeatureName => #{provided_by => rabbit,
                                       stability => stable}},
-    inject_on_nodes(Nodes, FeatureFlags),
+    ?assertEqual(ok, inject_on_nodes(Nodes, FeatureFlags)),
 
     ct:pal(
       "Checking the feature flag is supported but disabled on all nodes"),
@@ -469,7 +481,7 @@ enable_partially_supported_feature_flag_in_a_3node_cluster(Config) ->
     FeatureName = ?FUNCTION_NAME,
     FeatureFlags = #{FeatureName => #{provided_by => ?MODULE,
                                       stability => stable}},
-    inject_on_nodes([FirstNode], FeatureFlags),
+    ?assertEqual(ok, inject_on_nodes([FirstNode], FeatureFlags)),
 
     ct:pal(
       "Checking the feature flag is supported but disabled on all nodes"),
@@ -536,7 +548,7 @@ enable_unsupported_feature_flag_in_a_3node_cluster(Config) ->
     FeatureName = ?FUNCTION_NAME,
     FeatureFlags = #{FeatureName => #{provided_by => rabbit,
                                       stability => stable}},
-    inject_on_nodes([FirstNode], FeatureFlags),
+    ?assertEqual(ok, inject_on_nodes([FirstNode], FeatureFlags)),
 
     ct:pal(
       "Checking the feature flag is unsupported and disabled on all nodes"),
@@ -587,7 +599,7 @@ enable_feature_flag_in_cluster_and_add_member_after(Config) ->
                      #{provided_by => rabbit,
                        stability => stable,
                        callbacks => #{enable => {?MODULE, mf_count_runs}}}},
-    inject_on_nodes(AllNodes, FeatureFlags),
+    ?assertEqual(ok, inject_on_nodes(AllNodes, FeatureFlags)),
 
     ct:pal(
       "Checking the feature flag is supported but disabled on all nodes"),
@@ -691,7 +703,7 @@ enable_feature_flag_in_cluster_and_add_member_concurrently_mfv2(Config) ->
                        callbacks =>
                        #{enable =>
                          {?MODULE, mf_wait_and_count_runs_v2_enable}}}},
-    inject_on_nodes(AllNodes, FeatureFlags),
+    ?assertEqual(ok, inject_on_nodes(AllNodes, FeatureFlags)),
 
     ct:pal(
       "Checking the feature flag is supported but disabled on all nodes"),
@@ -876,7 +888,7 @@ enable_feature_flag_in_cluster_and_remove_member_concurrently_mfv2(Config) ->
                        callbacks =>
                        #{enable =>
                          {?MODULE, mf_wait_and_count_runs_v2_enable}}}},
-    inject_on_nodes(AllNodes, FeatureFlags),
+    ?assertEqual(ok, inject_on_nodes(AllNodes, FeatureFlags)),
 
     ct:pal(
       "Checking the feature flag is supported but disabled on all nodes"),
@@ -995,7 +1007,7 @@ enable_feature_flag_with_post_enable(Config) ->
                        callbacks =>
                        #{post_enable =>
                          {?MODULE, mf_wait_and_count_runs_v2_post_enable}}}},
-    inject_on_nodes(AllNodes, FeatureFlags),
+    ?assertEqual(ok, inject_on_nodes(AllNodes, FeatureFlags)),
 
     ct:pal(
       "Checking the feature flag is supported but disabled on all nodes"),
@@ -1180,8 +1192,8 @@ have_required_feature_flag_in_cluster_and_add_member_with_it_disabled(
     RequiredFeatureFlags = #{FeatureName =>
                              #{provided_by => rabbit,
                                stability => required}},
-    inject_on_nodes([NewNode], FeatureFlags),
-    inject_on_nodes(Nodes, RequiredFeatureFlags),
+    ?assertEqual(ok, inject_on_nodes([NewNode], FeatureFlags)),
+    ?assertEqual(ok, inject_on_nodes(Nodes, RequiredFeatureFlags)),
 
     ct:pal(
       "Checking the feature flag is supported everywhere but enabled on the "
@@ -1263,8 +1275,8 @@ have_required_feature_flag_in_cluster_and_add_member_without_it(
     RequiredFeatureFlags = #{FeatureName =>
                              #{provided_by => rabbit,
                                stability => required}},
-    inject_on_nodes([NewNode], FeatureFlags),
-    inject_on_nodes(Nodes, RequiredFeatureFlags),
+    ?assertEqual(ok, inject_on_nodes([NewNode], FeatureFlags)),
+    ?assertEqual(ok, inject_on_nodes(Nodes, RequiredFeatureFlags)),
 
     ct:pal(
       "Checking the feature flag is supported and enabled on existing the "
@@ -1359,7 +1371,7 @@ error_during_migration_after_initial_success(Config) ->
                        stability => stable,
                        callbacks =>
                        #{enable => {?MODULE, mf_crash_on_joining_node}}}},
-    inject_on_nodes(AllNodes, FeatureFlags),
+    ?assertEqual(ok, inject_on_nodes(AllNodes, FeatureFlags)),
 
     ct:pal(
       "Checking the feature flag is supported but disabled on all nodes"),