Support MQTT 5.0 features No Local, RAP, Subscription IDs

Support subscription options "No Local" and "Retain As Published"
as well as Subscription Identifiers.

All three MQTT 5.0 features can be set on a per subscription basis.
Due to wildcards in topic filters, multiple subscriptions
can match a given topic. Therefore, to implement Retain As Published and
Subscription Identifiers, the destination MQTT connection process needs
to know what subscription(s) caused it to receive the message.

There are a few ways how this could be implemented:

1. The destination MQTT connection process is aware of all its
   subscriptions. Whenever, it receives a message, it can match the
   message's routing key / topic against all its known topic filters.
   However, to iteratively match the routing key against all topic
   filters for every received message can become very expensive in the
   worst case when the MQTT client creates many subscriptions containing
   wildcards. This could be the case for an MQTT client that acts as a
   bridge or proxy or dispatcher: It could subscribe via a wildcard for
   each of its own clients.

2. Instead of interatively matching the topic of the received message
   against all topic filters that contain wildcards, a better approach
   would be for every MQTT subscriber connection process to maintain a
   local trie datastructure (similar to how topic exchanges are
   implemented) and perform matching therefore more efficiently.
   However, this does not sound optimal either because routing is
   effectively performed twice: in the topic exchange and again against
   a much smaller trie in each destination connection process.

3. Given that the topic exchange already perform routing, a much more
   sensible way would be to send the matched binding key(s) to the
   destination MQTT connection process. A subscription (topic filter)
   maps to a binding key in AMQP 0.9.1 routing. Therefore, for the first
   time in RabbitMQ, the routing function should not only output a list
   of unique destination queues, but also the binding keys (subscriptions)
   that caused the message to be routed to the destination queue.

This commit therefore implements the 3rd approach.
The downside of the 3rd approach is that it requires API changes to the
routing function and topic exchange.

Specifically, this commit adds a new function rabbit_exchange:route/3
that accepts a list of routing options. If that list contains version 2,
the caller of the routing function knows how to handle the return value
that could also contain binding keys.

This commits allows an MQTT connection process, the channel process, and
at-most-once dead lettering to handle binding keys. Binding keys are
included as AMQP 0.9.1 headers into the basic message.
Therefore, whenever a message is sent from an MQTT client or AMQP 0.9.1
client or AMQP 1.0 client or STOMP client, the MQTT receiver will know
the subscription identifier that caused the message to be received.

Note that due to the low number of allowed wildcard characters (# and
+), the cardinality of matched binding keys shouldn't be high even if
the topic contains for example 3 levels and the message is sent to for
example 5 million destination queues. In other words, sending multiple
distinct basic messages to the destination shouldn't hurt the delegate
optimisation too much. The delegate optimisation implemented for classic
queues and rabbit_mqtt_qos0_queue(s) still takes place for all basic
messages that contain the same set of matched binding keys.

The topic exchange returns all matched binding keys by remembering the
edges walked down to the leaves. As an optimisation, only for MQTT
queues are binding keys being returned. This does add a small dependency
from app rabbit to app rabbitmq_mqtt which is not optimal. However, this
dependency should be simple to remove when omitting this optimisation.

Another important feature of this commit is persisting subscription
options and subscription identifiers because they are part of the
MQTT 5.0 session state.

In MQTT v3 and v4, the only subscription information that were part of
the session state was the topic filter and the QoS level.
Both information were implicitly stored in the form of bindings:
The topic filter as the binding key and the QoS level as the destination
queue name of the binding.

For MQTT v5 we need to persist more subscription information.
From a domain perspective, it makes sense to store subscription options
as part of subscriptions, i.e. bindings, even though they are currently
not used in routing.
Therefore, this commits stores subscription options as binding arguments.

Storing subscription options as binding arguments comes in turn with
new challenges: How to handle mixed version clusters and upgrading an
MQTT session from v3 or v4 to v5?
Imagine an MQTT client connects via v5 with Session Expiry Interval > 0
to a new node in a mixed version cluster, creates a subscription,
disconnects, and subsequently connects via v3 to an old node. The
client should continue to receive messages.

To simplify such edge cases, this commit introduces a new feature flag
called mqtt_v5. If mqtt_v5 is disabled, clients cannot connect to
RabbitMQ via MQTT 5.0.

This still doesn't entirely solve the problem of MQTT session upgrades
(v4 to v5 client) or session downgrades (v5 to v4 client).

Ideally, once mqtt_v5 is enabled, all MQTT bindings contain non-empty binding
arguments. However, this will require a feature flag migration function
to modify all MQTT bindings. To be more precise, all MQTT bindings need
to be deleted and added because the binding argument is part of the
Mnesia table key.

Since feature flag migration functions are non-trivial to implement in
RabbitMQ (they can run on every node multiple times and concurrently),
this commit takes a simpler approach:
All v3 / v4 sessions keep the empty binding argument [].
All v5 sessions use the new binding argument [#mqtt_subscription_opts{}].

This requires only handling a session upgrade / downgrade by
creating a binding (with the new binding arg) and deleting the old
binding (with the old binding arg) when processing the CONNECT packet.

Note that such session upgrades or downgrades should be rather rare in
practice. Therefore these binding transactions shouldn't hurt peformance.

The No Local option is implemented within the MQTT publishing connection
process: The message is not sent to the MQTT destination if the
destination queue name matches the current MQTT client ID and the
message was routed due to a subscription that has the No Local flag set.
This avoids unnecessary traffic on the MQTT queue.
The alternative would have been that the "receiving side" (same process)
filters the message out - which would have been more consistent in how
Retain As Published and Subscription Identifiers are implemented, but
would have caused unnecessary load on the MQTT queue.

28 files changed, 1654 insertions, 311 deletions

diff --git a/deps/rabbit/src/rabbit_amqqueue.erl b/deps/rabbit/src/rabbit_amqqueue.erl
index d20d2a8da1..ca7a36ed5e 100644
--- a/deps/rabbit/src/rabbit_amqqueue.erl
+++ b/deps/rabbit/src/rabbit_amqqueue.erl
@@ -68,6 +68,7 @@
 -export([deactivate_limit_all/2]).
 
 -export([prepend_extra_bcc/1]).
+-export([queue/1, queue_name/1, queue_names/1]).
 
 %% internal
 -export([internal_declare/2, internal_delete/2, run_backing_queue/3,
@@ -101,6 +102,7 @@
 -type queue_not_found() :: not_found.
 -type queue_absent() :: {'absent', amqqueue:amqqueue(), absent_reason()}.
 -type not_found_or_absent() :: queue_not_found() | queue_absent().
+-type binding_keys() :: [rabbit_types:binding_key()].
 
 %%----------------------------------------------------------------------------
 
@@ -316,8 +318,8 @@ lookup(Name) when is_record(Name, resource) ->
 lookup_durable_queue(QName) ->
     rabbit_db_queue:get_durable(QName).
 
--spec lookup_many ([name()]) -> [amqqueue:amqqueue()].
-
+-spec lookup_many(rabbit_exchange:route_v2_destinations()) ->
+    [amqqueue:amqqueue() | {amqqueue:amqqueue(), binding_keys()}].
 lookup_many([])     -> [];                             %% optimisation
 lookup_many(Names) when is_list(Names) ->
     rabbit_db_queue:get_many(Names).
@@ -1993,11 +1995,13 @@ get_quorum_nodes(Q) ->
             []
     end.
 
--spec prepend_extra_bcc([amqqueue:amqqueue()]) ->
-    [amqqueue:amqqueue()].
+-spec prepend_extra_bcc(Qs) ->
+    Qs when Qs :: [amqqueue:amqqueue() |
+                   {amqqueue:amqqueue(), binding_keys()}].
 prepend_extra_bcc([]) ->
     [];
-prepend_extra_bcc([Q] = Qs) ->
+prepend_extra_bcc([Q0] = Qs) ->
+    Q = queue(Q0),
     case amqqueue:get_options(Q) of
         #{extra_bcc := BCCName} ->
             case get_bcc_queue(Q, BCCName) of
@@ -2012,7 +2016,8 @@ prepend_extra_bcc([Q] = Qs) ->
 prepend_extra_bcc(Qs) ->
     BCCQueues =
         lists:filtermap(
-          fun(Q) ->
+          fun(Q0) ->
+                  Q = queue(Q0),
                   case amqqueue:get_options(Q) of
                       #{extra_bcc := BCCName} ->
                           case get_bcc_queue(Q, BCCName) of
@@ -2027,6 +2032,34 @@ prepend_extra_bcc(Qs) ->
           end, Qs),
     lists:usort(BCCQueues) ++ Qs.
 
+-spec queue(Q | {Q, binding_keys()}) ->
+    Q when Q :: amqqueue:amqqueue().
+queue(Q)
+  when ?is_amqqueue(Q) ->
+    Q;
+queue({Q, BindingKeys})
+  when ?is_amqqueue(Q) andalso is_list(BindingKeys) ->
+    Q.
+
+-spec queue_name(name() | {name(), binding_keys()}) ->
+    name().
+queue_name(QName = #resource{kind = queue}) ->
+    QName;
+queue_name({QName = #resource{kind = queue}, BindingKeys})
+  when is_list(BindingKeys) ->
+    QName.
+
+-spec queue_names([Q | {Q, binding_keys()}]) ->
+    [name()] when Q :: amqqueue:amqqueue().
+queue_names(Queues)
+  when is_list(Queues) ->
+    lists:map(fun({Q, BindingKeys})
+                    when ?is_amqqueue(Q) andalso is_list(BindingKeys) ->
+                      amqqueue:get_name(Q);
+                 (Q) when ?is_amqqueue(Q) ->
+                      amqqueue:get_name(Q)
+              end, Queues).
+
 -spec get_bcc_queue(amqqueue:amqqueue(), binary()) ->
     {ok, amqqueue:amqqueue()} | {error, not_found}.
 get_bcc_queue(Q, BCCName) ->
diff --git a/deps/rabbit/src/rabbit_channel.erl b/deps/rabbit/src/rabbit_channel.erl
index fbf04f08b3..a9ac99ebed 100644
--- a/deps/rabbit/src/rabbit_channel.erl
+++ b/deps/rabbit/src/rabbit_channel.erl
@@ -1306,7 +1306,7 @@ handle_method(#'basic.publish'{exchange    = ExchangeNameBin,
         {ok, Message} ->
             Delivery = rabbit_basic:delivery(
                          Mandatory, DoConfirm, Message, MsgSeqNo),
-            QNames = rabbit_exchange:route(Exchange, Delivery),
+            QNames = rabbit_exchange:route(Exchange, Delivery, [v2]),
             rabbit_trace:tap_in(Message, QNames, ConnName, ChannelNum,
                                 Username, TraceState),
             DQ = {Delivery#delivery{flow = Flow}, QNames},
@@ -2157,13 +2157,13 @@ deliver_to_queues({Delivery = #delivery{message    = Message = #basic_message{ex
                    RoutedToQueueNames = [QName]}, State0 = #ch{queue_states = QueueStates0}) -> %% optimisation when there is one queue
     Qs0 = rabbit_amqqueue:lookup_many(RoutedToQueueNames),
     Qs = rabbit_amqqueue:prepend_extra_bcc(Qs0),
-    QueueNames = lists:map(fun amqqueue:get_name/1, Qs),
     case rabbit_queue_type:deliver(Qs, Delivery, QueueStates0) of
         {ok, QueueStates, Actions}  ->
             rabbit_global_counters:messages_routed(amqp091, erlang:min(1, length(Qs))),
             %% NB: the order here is important since basic.returns must be
             %% sent before confirms.
             ok = process_routing_mandatory(Mandatory, Qs, Message, State0),
+            QueueNames = rabbit_amqqueue:queue_names(Qs),
             State1 = process_routing_confirm(Confirm, QueueNames, MsgSeqNo, XName, State0),
             %% Actions must be processed after registering confirms as actions may
             %% contain rejections of publishes
@@ -2171,7 +2171,7 @@ deliver_to_queues({Delivery = #delivery{message    = Message = #basic_message{ex
             case rabbit_event:stats_level(State, #ch.stats_timer) of
                 fine ->
                     ?INCR_STATS(exchange_stats, XName, 1, publish),
-                    ?INCR_STATS(queue_exchange_stats, {QName, XName}, 1, publish);
+                    ?INCR_STATS(queue_exchange_stats, {rabbit_amqqueue:queue_name(QName), XName}, 1, publish);
                 _ ->
                     ok
             end,
@@ -2194,13 +2194,13 @@ deliver_to_queues({Delivery = #delivery{message    = Message = #basic_message{ex
                    RoutedToQueueNames}, State0 = #ch{queue_states = QueueStates0}) ->
     Qs0 = rabbit_amqqueue:lookup_many(RoutedToQueueNames),
     Qs = rabbit_amqqueue:prepend_extra_bcc(Qs0),
-    QueueNames = lists:map(fun amqqueue:get_name/1, Qs),
     case rabbit_queue_type:deliver(Qs, Delivery, QueueStates0) of
         {ok, QueueStates, Actions}  ->
             rabbit_global_counters:messages_routed(amqp091, length(Qs)),
             %% NB: the order here is important since basic.returns must be
             %% sent before confirms.
             ok = process_routing_mandatory(Mandatory, Qs, Message, State0),
+            QueueNames = rabbit_amqqueue:queue_names(Qs),
             State1 = process_routing_confirm(Confirm, QueueNames,
                                              MsgSeqNo, XName, State0),
             %% Actions must be processed after registering confirms as actions may
diff --git a/deps/rabbit/src/rabbit_db_queue.erl b/deps/rabbit/src/rabbit_db_queue.erl
index 032416b02d..b72925383b 100644
--- a/deps/rabbit/src/rabbit_db_queue.erl
+++ b/deps/rabbit/src/rabbit_db_queue.erl
@@ -326,20 +326,36 @@ internal_delete_in_mnesia(QueueName, OnlyDurable, Reason) ->
 %% get_many().
 %% -------------------------------------------------------------------
 
--spec get_many([QName]) -> Ret when
-      QName :: rabbit_amqqueue:name(),
-      Ret :: [Queue :: amqqueue:amqqueue()].
+-spec get_many(rabbit_exchange:route_v2_destinations()) ->
+    [amqqueue:amqqueue() | {amqqueue:amqqueue(), [rabbit_types:binding_key(), ...]}].
 get_many(Names) when is_list(Names) ->
     rabbit_db:run(
       #{mnesia => fun() -> get_many_in_mnesia(?MNESIA_TABLE, Names) end
        }).
-
+get_many_in_mnesia(Table, [{Name, BindingKeys}])
+  when is_list(BindingKeys) ->
+    case ets:lookup(Table, Name) of
+        [] -> [];
+        [Q] -> [{Q, BindingKeys}]
+    end;
 get_many_in_mnesia(Table, [Name]) ->
     ets:lookup(Table, Name);
-get_many_in_mnesia(Table, Names) when is_list(Names) ->
+get_many_in_mnesia(Table, Names)
+  when is_list(Names) ->
     %% Normally we'd call mnesia:dirty_read/1 here, but that is quite
     %% expensive for reasons explained in rabbit_mnesia:dirty_read/1.
-    lists:append([ets:lookup(Table, Name) || Name <- Names]).
+    lists:filtermap(fun({Name, BindingKeys})
+                          when is_list(BindingKeys) ->
+                            case ets:lookup(Table, Name) of
+                                [] -> false;
+                                [Q] -> {true, {Q, BindingKeys}}
+                            end;
+                       (Name) ->
+                            case ets:lookup(Table, Name) of
+                                [] -> false;
+                                [Q] -> {true, Q}
+                            end
+                    end, Names).
 
 %% -------------------------------------------------------------------
 %% get().
diff --git a/deps/rabbit/src/rabbit_db_topic_exchange.erl b/deps/rabbit/src/rabbit_db_topic_exchange.erl
index 5c0f0283ef..493442c350 100644
--- a/deps/rabbit/src/rabbit_db_topic_exchange.erl
+++ b/deps/rabbit/src/rabbit_db_topic_exchange.erl
@@ -9,7 +9,7 @@
 
 -include_lib("rabbit_common/include/rabbit.hrl").
 
--export([set/1, delete_all_for_exchange/1, delete/1, match/2]).
+-export([set/1, delete_all_for_exchange/1, delete/1, match/3]).
 
 %% For testing
 -export([clear/0]).
@@ -18,6 +18,9 @@
 -define(MNESIA_EDGE_TABLE, rabbit_topic_trie_edge).
 -define(MNESIA_BINDING_TABLE, rabbit_topic_trie_binding).
 
+-type match_ret() :: {[rabbit_types:binding_destination()],
+                      #{rabbit_amqqueue:name() => [rabbit_types:binding_key(), ...]}}.
+
 %% -------------------------------------------------------------------
 %% set().
 %% -------------------------------------------------------------------
@@ -67,21 +70,21 @@ delete(Bs) when is_list(Bs) ->
 %% match().
 %% -------------------------------------------------------------------
 
--spec match(ExchangeName, RoutingKey) -> ok when
-      ExchangeName :: rabbit_exchange:name(),
-      RoutingKey :: binary().
-%% @doc Finds the topic binding matching the given exchange and routing key and returns
-%% the destination of the binding
+-spec match(rabbit_exchange:name(),
+            rabbit_types:routing_key(),
+            rabbit_exchange:route_opts()) -> match_ret().
+%% @doc Finds the topic bindings matching the given exchange and routing key and returns
+%% the destination of the bindings (with the matched bindings in v2).
 %%
-%% @returns a list of resources
+%% @returns a list of resources (with matched bindings in v2).
 %%
 %% @private
 
-match(XName, RoutingKey) ->
+match(XName, RoutingKey, Opts) ->
     rabbit_db:run(
       #{mnesia =>
             fun() ->
-                    match_in_mnesia(XName, RoutingKey)
+                    match_in_mnesia(XName, RoutingKey, lists:member(v2, Opts))
             end
        }).
 
@@ -128,9 +131,9 @@ delete_all_for_exchange_in_mnesia(XName) ->
               ok
       end).
 
-match_in_mnesia(XName, RoutingKey) ->
+match_in_mnesia(XName, RoutingKey, V2) ->
     Words = split_topic_key(RoutingKey),
-    mnesia:async_dirty(fun trie_match/2, [XName, Words]).
+    mnesia:async_dirty(fun trie_match/3, [XName, Words, V2]).
 
 trie_remove_all_nodes(X) ->
     remove_all(?MNESIA_NODE_TABLE,
@@ -188,30 +191,32 @@ split_topic_key(<<$., Rest/binary>>, RevWordAcc, RevResAcc) ->
 split_topic_key(<<C:8, Rest/binary>>, RevWordAcc, RevResAcc) ->
     split_topic_key(Rest, [C | RevWordAcc], RevResAcc).
 
-trie_match(X, Words) ->
-    trie_match(X, root, Words, []).
+trie_match(X, Words, V2) ->
+    trie_match(X, root, Words, V2, [], {[], #{}}).
 
-trie_match(X, Node, [], ResAcc) ->
-    trie_match_part(X, Node, "#", fun trie_match_skip_any/4, [],
-                    trie_bindings(X, Node) ++ ResAcc);
-trie_match(X, Node, [W | RestW] = Words, ResAcc) ->
+trie_match(X, Node, [], V2, Walked, ResAcc0) ->
+    Destinations = trie_bindings(X, Node),
+    ResAcc = maybe_add_binding_key(V2, Destinations, Walked, ResAcc0),
+    trie_match_part(X, Node, "#", fun trie_match_skip_any/6, [],
+                    V2, Walked, ResAcc);
+trie_match(X, Node, [W | RestW] = Words, V2, Walked, ResAcc) ->
     lists:foldl(fun ({WArg, MatchFun, RestWArg}, Acc) ->
-                        trie_match_part(X, Node, WArg, MatchFun, RestWArg, Acc)
-                end, ResAcc, [{W, fun trie_match/4, RestW},
-                              {"*", fun trie_match/4, RestW},
-                              {"#", fun trie_match_skip_any/4, Words}]).
+                        trie_match_part(X, Node, WArg, MatchFun, RestWArg, V2, Walked, Acc)
+                end, ResAcc, [{W, fun trie_match/6, RestW},
+                              {"*", fun trie_match/6, RestW},
+                              {"#", fun trie_match_skip_any/6, Words}]).
 
-trie_match_part(X, Node, Search, MatchFun, RestW, ResAcc) ->
+trie_match_part(X, Node, Search, MatchFun, RestW, V2, Walked, ResAcc) ->
     case trie_child(X, Node, Search) of
-        {ok, NextNode} -> MatchFun(X, NextNode, RestW, ResAcc);
+        {ok, NextNode} -> MatchFun(X, NextNode, RestW, V2, [Search | Walked], ResAcc);
         error          -> ResAcc
     end.
 
-trie_match_skip_any(X, Node, [], ResAcc) ->
-    trie_match(X, Node, [], ResAcc);
-trie_match_skip_any(X, Node, [_ | RestW] = Words, ResAcc) ->
-    trie_match_skip_any(X, Node, RestW,
-                        trie_match(X, Node, Words, ResAcc)).
+trie_match_skip_any(X, Node, [], V2, Walked, ResAcc) ->
+    trie_match(X, Node, [], V2, Walked, ResAcc);
+trie_match_skip_any(X, Node, [_ | RestW] = Words, V2, Walked, ResAcc) ->
+    trie_match_skip_any(X, Node, RestW, V2, Walked,
+                        trie_match(X, Node, Words, V2, Walked, ResAcc)).
 
 follow_down_create(X, Words) ->
     case follow_down_last_node(X, Words) of
@@ -323,3 +328,35 @@ trie_binding_op(X, Node, D, Args, Op) ->
                                            destination   = D,
                                            arguments     = Args}},
             write).
+
+-spec maybe_add_binding_key(boolean(), [rabbit_types:binding_destination()], [Word :: [byte()]], Acc) ->
+    Acc when Acc :: match_ret().
+maybe_add_binding_key(false, Destinations, _, {DestinationsAcc, QNamesToBindingsAcc}) ->
+    {Destinations ++ DestinationsAcc, QNamesToBindingsAcc};
+maybe_add_binding_key(true, Destinations, Walked, Acc = {DestinationsAcc, QNamesToBindingsAcc}) ->
+    %% This is the only place where app rabbit is aware of app rabbitmq_mqtt.
+    %% Ideally, such a dependency does not exist. In fact, removing MQTT specifics below
+    %% (i.e. feature flag and queue name prefix) works as well. Checking for MQTT
+    %% destination queues is just an optimisation to not unnecessarily add an AMQP 0.9.1
+    %% header for messages sent to other destinations.
+    case rabbit_feature_flags:is_enabled(mqtt_v5) of
+        false ->
+            {Destinations ++ DestinationsAcc, QNamesToBindingsAcc};
+        true ->
+            %% TODO Benchmark. If binary creation is too expensive, implement one of alternatives:
+            %% 1. Create binary lazily only if Destinations include at least one MQTT queue
+            %% 2. Return binding args in trie_bindings/2 and create binary lazily if binding args
+            %%    have Subscription Identifier or Retain As Published flag set.
+            %% 3. Store subscriptions by lists instead of by binary in rabbit_mqtt_processor.erl
+            BindingKey = list_to_binary(lists:join($., lists:reverse(Walked))),
+            lists:foldl(
+              fun(QName = #resource{kind = queue,
+                                    name = <<"mqtt-subscription-", _ClientIdQos/binary>>},
+                  {A0, A1}) ->
+                      {A0, maps:update_with(QName, fun(BindingKeys) ->
+                                                           [BindingKey | BindingKeys]
+                                                   end, [BindingKey], A1)};
+                 (Destination, {A0, A1}) ->
+                      {[Destination | A0], A1}
+              end, Acc, Destinations)
+    end.
diff --git a/deps/rabbit/src/rabbit_dead_letter.erl b/deps/rabbit/src/rabbit_dead_letter.erl
index 8baa94774f..679ce7edbf 100644
--- a/deps/rabbit/src/rabbit_dead_letter.erl
+++ b/deps/rabbit/src/rabbit_dead_letter.erl
@@ -26,8 +26,8 @@
 publish(Msg, Reason, X, RK, SourceQName) ->
     DLMsg = make_msg(Msg, Reason, X#exchange.name, RK, SourceQName),
     Delivery = rabbit_basic:delivery(false, false, DLMsg, undefined),
-    {QNames, Cycles} = detect_cycles(Reason, DLMsg,
-                                     rabbit_exchange:route(X, Delivery)),
+    QNames0 = rabbit_exchange:route(X, Delivery, [v2]),
+    {QNames, Cycles} = detect_cycles(Reason, DLMsg, QNames0),
     lists:foreach(fun log_cycle_once/1, Cycles),
     Qs0 = rabbit_amqqueue:lookup_many(QNames),
     Qs = rabbit_amqqueue:prepend_extra_bcc(Qs0),
@@ -209,13 +209,19 @@ detect_cycles(_Reason, #basic_message{content = Content}, Queues) ->
                 {array, Deaths} ->
                     {Cycling, NotCycling} =
                         lists:partition(fun (#resource{name = Queue}) ->
+                                                is_cycle(Queue, Deaths);
+                                            ({#resource{name = Queue}, _BindingKeys}) ->
                                                 is_cycle(Queue, Deaths)
                                         end, Queues),
                     OldQueues = [rabbit_misc:table_lookup(D, <<"queue">>) ||
                                     {table, D} <- Deaths],
                     OldQueues1 = [QName || {longstr, QName} <- OldQueues],
-                    {NotCycling, [[QName | OldQueues1] ||
-                                     #resource{name = QName} <- Cycling]};
+                    Cycling1 = lists:map(fun(#resource{name = QName}) ->
+                                                 [QName | OldQueues1];
+                                            ({#resource{name = QName}, _BindingKeys}) ->
+                                                 [QName | OldQueues1]
+                                         end, Cycling),
+                    {NotCycling, Cycling1};
                 _ ->
                     NoCycles
             end
diff --git a/deps/rabbit/src/rabbit_exchange.erl b/deps/rabbit/src/rabbit_exchange.erl
index 1e118f7c4a..79f23d021f 100644
--- a/deps/rabbit/src/rabbit_exchange.erl
+++ b/deps/rabbit/src/rabbit_exchange.erl
@@ -14,16 +14,20 @@
          lookup/1, lookup_many/1, lookup_or_die/1, list/0, list/1, lookup_scratch/2,
          update_scratch/3, update_decorators/2, immutable/1,
          info_keys/0, info/1, info/2, info_all/1, info_all/2, info_all/4,
-         route/2, delete/3, validate_binding/2, count/0]).
+         route/2, route/3, delete/3, validate_binding/2, count/0]).
 -export([list_names/0]).
 -export([serialise_events/1]).
 -export([serial/1, peek_serial/1]).
 
 %%----------------------------------------------------------------------------
 
--export_type([name/0, type/0]).
+-export_type([name/0, type/0, route_opts/0, route_v2_destinations/0]).
 -type name() :: rabbit_types:exchange_name().
 -type type() :: rabbit_types:exchange_type().
+-type route_opts() :: [v2].
+-type route_v2_destinations() :: [rabbit_amqqueue:name() |
+                                  {rabbit_amqqueue:name(),
+                                   [MatchedBindingKeys :: rabbit_types:binding_key(), ...]}].
 -type fun_name() :: atom().
 
 %%----------------------------------------------------------------------------
@@ -342,12 +346,16 @@ info_all(VHostPath, Items, Ref, AggregatorPid) ->
 %% fields can't be undefined. But there are places where
 %% rabbit_exchange:route/2 is called with the absolutely bare delivery
 %% like #delivery{message = #basic_message{routing_keys = [...]}}
--spec route(rabbit_types:exchange(), #delivery{})
-                 -> [rabbit_amqqueue:name()].
+-spec route(rabbit_types:exchange(), #delivery{}) -> [rabbit_amqqueue:name()].
+route(Exchange, Delivery) ->
+    route(Exchange, Delivery, []).
 
+-spec route (rabbit_types:exchange(), #delivery{}, route_opts()) ->
+    route_v2_destinations().
 route(#exchange{name = #resource{virtual_host = VHost, name = RName} = XName,
                 decorators = Decorators} = X,
-      #delivery{message = #basic_message{routing_keys = RKs}} = Delivery) ->
+      #delivery{message = #basic_message{routing_keys = RKs}} = Delivery,
+      Opts) ->
     case RName of
         <<>> ->
             RKsSorted = lists:usort(RKs),
@@ -357,22 +365,30 @@ route(#exchange{name = #resource{virtual_host = VHost, name = RName} = XName,
                                                 not virtual_reply_queue(RK)];
         _ ->
             Decs = rabbit_exchange_decorator:select(route, Decorators),
-            lists:usort(route1(Delivery, Decs, {[X], XName, []}))
+            lists:usort(route1(Delivery, Decs, Opts, {[X], XName, []}))
     end.
 
 virtual_reply_queue(<<"amq.rabbitmq.reply-to.", _/binary>>) -> true;
 virtual_reply_queue(_)                                      -> false.
 
-route1(_, _, {[], _, QNames}) ->
+route1(_, _, _, {[], _, QNames}) ->
     QNames;
-route1(Delivery, Decorators,
+route1(Delivery, Decorators, Opts,
        {[X = #exchange{type = Type} | WorkList], SeenXs, QNames}) ->
-    ExchangeDests  = (type_to_module(Type)):route(X, Delivery),
+    XMod = type_to_module(Type),
+    %% TODO benchmark cost of erlang:function_exported/3
+    %% If it's slow, only call route/3 if XMod =:= rabbit_exchange_type_topic
+    ExchangeDests = case erlang:function_exported(XMod, route, 3) of
+                        true ->
+                            XMod:route(X, Delivery, Opts);
+                        false ->
+                            XMod:route(X, Delivery)
+                    end,
     DecorateDests  = process_decorators(X, Decorators, Delivery),
     AlternateDests = process_alternate(X, ExchangeDests),
-    route1(Delivery, Decorators,
+    route1(Delivery, Decorators, Opts,
            lists:foldl(fun process_route/2, {WorkList, SeenXs, QNames},
-                       AlternateDests ++ DecorateDests  ++ ExchangeDests)).
+                       AlternateDests ++ DecorateDests ++ ExchangeDests)).
 
 process_alternate(X = #exchange{name = XName}, []) ->
     case rabbit_policy:get_arg(
@@ -404,7 +420,10 @@ process_route(#resource{kind = exchange} = XName,
     end;
 process_route(#resource{kind = queue} = QName,
               {WorkList, SeenXs, QNames}) ->
-    {WorkList, SeenXs, [QName | QNames]}.
+    {WorkList, SeenXs, [QName | QNames]};
+process_route({#resource{kind = queue} = QName, BindingKeys},
+              {WorkList, SeenXs, QNames}) ->
+    {WorkList, SeenXs, [{QName, lists:usort(BindingKeys)} | QNames]}.
 
 cons_if_present(XName, L) ->
     case lookup(XName) of
diff --git a/deps/rabbit/src/rabbit_exchange_type.erl b/deps/rabbit/src/rabbit_exchange_type.erl
index ecf62729b6..bfbf7aa7e0 100644
--- a/deps/rabbit/src/rabbit_exchange_type.erl
+++ b/deps/rabbit/src/rabbit_exchange_type.erl
@@ -26,6 +26,10 @@
 -callback route(rabbit_types:exchange(), rabbit_types:delivery()) ->
     rabbit_router:match_result().
 
+-callback route(rabbit_types:exchange(), rabbit_types:delivery(), rabbit_exchange:route_opts()) ->
+    [rabbit_types:binding_destination() |
+     {rabbit_amqqueue:name(), [rabbit_types:binding_key(), ...]}].
+
 %% called BEFORE declaration, to check args etc; may exit with #amqp_error{}
 -callback validate(rabbit_types:exchange()) -> 'ok'.
 
@@ -63,5 +67,7 @@
 
 -callback info(rabbit_types:exchange(), [atom()]) -> [{atom(), term()}].
 
+-optional_callbacks([route/3]).
+
 added_to_rabbit_registry(_Type, _ModuleName) -> ok.
 removed_from_rabbit_registry(_Type) -> ok.
diff --git a/deps/rabbit/src/rabbit_exchange_type_topic.erl b/deps/rabbit/src/rabbit_exchange_type_topic.erl
index ee94022b5f..b576645786 100644
--- a/deps/rabbit/src/rabbit_exchange_type_topic.erl
+++ b/deps/rabbit/src/rabbit_exchange_type_topic.erl
@@ -11,7 +11,7 @@
 
 -behaviour(rabbit_exchange_type).
 
--export([description/0, serialise_events/0, route/2]).
+-export([description/0, serialise_events/0, route/2, route/3]).
 -export([validate/1, validate_binding/2,
          create/2, delete/2, policy_changed/2, add_binding/3,
          remove_bindings/3, assert_args_equivalence/2]).
@@ -34,10 +34,32 @@ description() ->
 
 serialise_events() -> false.
 
-%% NB: This may return duplicate results in some situations (that's ok)
+route(Exchange, Delivery) ->
+    route(Exchange, Delivery, []).
+
+%% This function can return duplicate destinations or duplicate binding keys.
+%% The caller of this function is responsible for filtering out duplicates.
+-spec route(rabbit_types:exchange(), rabbit_types:delivery(), rabbit_exchange:route_opts()) ->
+    [rabbit_types:binding_destination() |
+     {rabbit_amqqueue:name(), [rabbit_types:binding_key(), ...]}].
 route(#exchange{name = XName},
-      #delivery{message = #basic_message{routing_keys = Routes}}) ->
-    lists:append([rabbit_db_topic_exchange:match(XName, RKey) || RKey <- Routes]).
+      #delivery{message = #basic_message{routing_keys = Routes}},
+      Opts) ->
+    {Destinations, QNamesWithBindings} =
+    lists:foldl(fun(RKey, {L0, M0}) ->
+                        {L1, M1} = rabbit_db_topic_exchange:match(XName, RKey, Opts),
+                        L = L0 ++ L1,
+                        M = maps:merge_with(fun(_QName, BindingKeys0, BindingKeys1) ->
+                                                    BindingKeys0 ++ BindingKeys1
+                                            end, M0, M1),
+                        {L, M}
+                end, {[], #{}}, Routes),
+    if map_size(QNamesWithBindings) =:= 0 ->
+           %% optimisation
+           Destinations;
+       true ->
+           Destinations ++ maps:to_list(QNamesWithBindings)
+    end.
 
 validate(_X) -> ok.
 validate_binding(_X, _B) -> ok.
diff --git a/deps/rabbit/src/rabbit_queue_type.erl b/deps/rabbit/src/rabbit_queue_type.erl
index 080ef50bfc..f4f9c7eff3 100644
--- a/deps/rabbit/src/rabbit_queue_type.erl
+++ b/deps/rabbit/src/rabbit_queue_type.erl
@@ -7,7 +7,7 @@
 
 -module(rabbit_queue_type).
 -include("amqqueue.hrl").
--include_lib("rabbit_common/include/resource.hrl").
+-include_lib("rabbit_common/include/rabbit.hrl").
 
 -export([
          init/0,
@@ -482,7 +482,9 @@ module(QRef, State) ->
             {error, not_found}
     end.
 
--spec deliver([amqqueue:amqqueue()], Delivery :: term(),
+-spec deliver([amqqueue:amqqueue() |
+               {amqqueue:amqqueue(), [rabbit_types:binding_key(), ...]}],
+              rabbit_types:delivery(),
               stateless | state()) ->
     {ok, state(), actions()} | {error, Reason :: term()}.
 deliver(Qs, Delivery, State) ->
@@ -493,41 +495,48 @@ deliver(Qs, Delivery, State) ->
             {error, Reason}
     end.
 
-deliver0(Qs, Delivery, stateless) ->
-    ByType = lists:foldl(fun(Q, Acc) ->
-                                 Mod = amqqueue:get_type(Q),
-                                 maps:update_with(
-                                   Mod, fun(A) ->
+deliver0(Qs, Delivery0, stateless) ->
+    ByTypeAndBindingKeys =
+    lists:foldl(fun(Elem, Acc) ->
+                        {Q, BKeys} = queue_and_binding_keys(Elem),
+                        Mod = amqqueue:get_type(Q),
+                        maps:update_with(
+                          {Mod, BKeys}, fun(A) ->
                                                 [{Q, stateless} | A]
                                         end, [{Q, stateless}], Acc)
-                         end, #{}, Qs),
-    maps:foreach(fun(Mod, QSs) ->
+                end, #{}, Qs),
+    maps:foreach(fun({Mod, BKeys}, QSs) ->
+                         Delivery = add_binding_keys(Delivery0, BKeys),
                          _ = Mod:deliver(QSs, Delivery)
-                 end, ByType),
+                 end, ByTypeAndBindingKeys),
     {ok, stateless, []};
-deliver0(Qs, Delivery, #?STATE{} = State0) ->
+deliver0(Qs, Delivery0, #?STATE{} = State0) ->
     %% TODO: optimise single queue case?
     %% sort by queue type - then dispatch each group
-    ByType = lists:foldl(
-               fun (Q, Acc) ->
-                       Mod = amqqueue:get_type(Q),
-                       QState = case Mod:is_stateful() of
-                                    true ->
-                                        #ctx{state = S} = get_ctx(Q, State0),
-                                        S;
-                                    false ->
-                                        stateless
-                                end,
-                       maps:update_with(
-                         Mod, fun (A) ->
+    ByTypeAndBindingKeys =
+    lists:foldl(
+      fun (Elem, Acc) ->
+              {Q, BKeys} = queue_and_binding_keys(Elem),
+              Mod = amqqueue:get_type(Q),
+              QState = case Mod:is_stateful() of
+                           true ->
+                               #ctx{state = S} = get_ctx(Q, State0),
+                               S;
+                           false ->
+                               stateless
+                       end,
+              maps:update_with(
+                {Mod, BKeys}, fun (A) ->
                                       [{Q, QState} | A]
                               end, [{Q, QState}], Acc)
-               end, #{}, Qs),
+      end, #{}, Qs),
     %%% dispatch each group to queue type interface?
-    {Xs, Actions} = maps:fold(fun(Mod, QSs, {X0, A0}) ->
-                                      {X, A} = Mod:deliver(QSs, Delivery),
-                                      {X0 ++ X, A0 ++ A}
-                              end, {[], []}, ByType),
+    {Xs, Actions} = maps:fold(
+                      fun({Mod, BKeys}, QSs, {X0, A0}) ->
+                              Delivery = add_binding_keys(Delivery0, BKeys),
+                              {X, A} = Mod:deliver(QSs, Delivery),
+                              {X0 ++ X, A0 ++ A}
+                      end, {[], []}, ByTypeAndBindingKeys),
     State = lists:foldl(
               fun({Q, S}, Acc) ->
                       Ctx = get_ctx_with(Q, Acc, S),
@@ -535,6 +544,25 @@ deliver0(Qs, Delivery, #?STATE{} = State0) ->
               end, State0, Xs),
     {ok, State, Actions}.
 
+queue_and_binding_keys(Q)
+  when ?is_amqqueue(Q) ->
+    {Q, []};
+queue_and_binding_keys(T = {Q, BindingKeys})
+  when ?is_amqqueue(Q) andalso is_list(BindingKeys) ->
+    T.
+
+-spec add_binding_keys(rabbit_types:delivery(), [rabbit_types:binding_key()]) ->
+    rabbit_types:delivery().
+add_binding_keys(Delivery, []) ->
+    Delivery;
+add_binding_keys(Delivery, BindingKeys) ->
+    L = [{longstr, B} || B <- BindingKeys],
+    BasicMsg = rabbit_basic:add_header(
+                 %% TODO avoid binary creation?
+                 <<"x-binding-keys">>, array, L,
+                 Delivery#delivery.message),
+    Delivery#delivery{message = BasicMsg}.
+
 -spec settle(queue_name(), settle_op(), rabbit_types:ctag(),
              [non_neg_integer()], state()) ->
           {ok, state(), actions()} |
diff --git a/deps/rabbit/src/rabbit_trace.erl b/deps/rabbit/src/rabbit_trace.erl
index 6c55630137..57c5393f9d 100644
--- a/deps/rabbit/src/rabbit_trace.erl
+++ b/deps/rabbit/src/rabbit_trace.erl
@@ -47,12 +47,12 @@ enabled(none) ->
 enabled(#exchange{}) ->
     true.
 
--spec tap_in(rabbit_types:basic_message(), [rabbit_amqqueue:name()],
+-spec tap_in(rabbit_types:basic_message(), rabbit_exchange:route_v2_destinations(),
              binary(), rabbit_types:username(), state()) -> 'ok'.
 tap_in(Msg, QNames, ConnName, Username, State) ->
     tap_in(Msg, QNames, ConnName, ?CONNECTION_GLOBAL_CHANNEL_NUM, Username, State).
 
--spec tap_in(rabbit_types:basic_message(), [rabbit_amqqueue:name()],
+-spec tap_in(rabbit_types:basic_message(), rabbit_exchange:route_v2_destinations(),
                    binary(), rabbit_channel:channel_number(),
                    rabbit_types:username(), state()) -> 'ok'.
 
@@ -60,13 +60,17 @@ tap_in(_Msg, _QNames, _ConnName, _ChannelNum, _Username, none) -> ok;
 tap_in(Msg = #basic_message{exchange_name = #resource{name         = XName,
                                                       virtual_host = VHost}},
        QNames, ConnName, ChannelNum, Username, TraceX) ->
+    RoutedQs = lists:map(fun(#resource{kind = queue, name = Name}) ->
+                                 {longstr, Name};
+                            ({#resource{kind = queue, name = Name}, _BindingKeys}) ->
+                                 {longstr, Name}
+                         end, QNames),
     trace(TraceX, Msg, <<"publish">>, XName,
           [{<<"vhost">>,         longstr,   VHost},
            {<<"connection">>,    longstr,   ConnName},
            {<<"channel">>,       signedint, ChannelNum},
            {<<"user">>,          longstr,   Username},
-           {<<"routed_queues">>, array,
-            [{longstr, QName#resource.name} || QName <- QNames]}]).
+           {<<"routed_queues">>, array,     RoutedQs}]).
 
 -spec tap_out(rabbit_amqqueue:qmsg(), binary(),
               rabbit_types:username(), state()) -> 'ok'.
diff --git a/deps/rabbit/test/rabbit_db_topic_exchange_SUITE.erl b/deps/rabbit/test/rabbit_db_topic_exchange_SUITE.erl
index 8c2d424e20..85af7f2981 100644
--- a/deps/rabbit/test/rabbit_db_topic_exchange_SUITE.erl
+++ b/deps/rabbit/test/rabbit_db_topic_exchange_SUITE.erl
@@ -9,20 +9,19 @@
 
 -include_lib("rabbit_common/include/rabbit.hrl").
 -include_lib("eunit/include/eunit.hrl").
--include_lib("common_test/include/ct.hrl").
 
--compile(export_all).
+-compile([nowarn_export_all, export_all]).
 
 -define(VHOST, <<"/">>).
 
 all() ->
     [
-     {group, all_tests}
+     {group, tests}
     ].
 
 groups() ->
     [
-     {all_tests, [], all_tests()}
+     {tests, [shuffle], all_tests()}
     ].
 
 all_tests() ->
@@ -30,7 +29,9 @@ all_tests() ->
      set,
      delete,
      delete_all_for_exchange,
-     match
+     match_v1,
+     match_v2_many_destinations,
+     match_v2_single_destination
     ].
 
 %% -------------------------------------------------------------------
@@ -77,10 +78,10 @@ set1(_Config) ->
     Dst = rabbit_misc:r(?VHOST, queue, <<"test-queue">>),
     RoutingKey = <<"a.b.c">>,
     Binding = #binding{source = Src, key = RoutingKey, destination = Dst, args = #{}},
-    ?assertEqual([], rabbit_db_topic_exchange:match(Src, RoutingKey)),
+    ?assertEqual({[], #{}}, rabbit_db_topic_exchange:match(Src, RoutingKey, [])),
     ?assertEqual(ok, rabbit_db_topic_exchange:set(Binding)),
     ?assertEqual(ok, rabbit_db_topic_exchange:set(Binding)),
-    ?assertEqual([Dst], rabbit_db_topic_exchange:match(Src, RoutingKey)),
+    ?assertEqual({[Dst], #{}}, rabbit_db_topic_exchange:match(Src, RoutingKey, [])),
     passed.
 
 delete(Config) ->
@@ -100,9 +101,10 @@ delete1(_Config) ->
     ?assertEqual(ok, rabbit_db_topic_exchange:set(Binding1)),
     ?assertEqual(ok, rabbit_db_topic_exchange:set(Binding2)),
     ?assertEqual(ok, rabbit_db_topic_exchange:set(Binding3)),
-    ?assertEqual(Dsts, lists:sort(rabbit_db_topic_exchange:match(Src, RoutingKey))),
+    {Actual, #{}} = rabbit_db_topic_exchange:match(Src, RoutingKey, []),
+    ?assertEqual(Dsts, lists:sort(Actual)),
     ?assertEqual(ok, rabbit_db_topic_exchange:delete([Binding1, Binding2])),
-    ?assertEqual([Dst3], rabbit_db_topic_exchange:match(Src, RoutingKey)),
+    ?assertEqual({[Dst3], #{}}, rabbit_db_topic_exchange:match(Src, RoutingKey, [])),
     passed.
 
 delete_all_for_exchange(Config) ->
@@ -119,17 +121,18 @@ delete_all_for_exchange1(_Config) ->
     set(Src1, RoutingKey, Dst1),
     set(Src1, RoutingKey, Dst2),
     set(Src2, RoutingKey, Dst1),
-    ?assertEqual(Dsts, lists:sort(rabbit_db_topic_exchange:match(Src1, RoutingKey))),
-    ?assertEqual([Dst1], rabbit_db_topic_exchange:match(Src2, RoutingKey)),
+    {Actual, #{}} = rabbit_db_topic_exchange:match(Src1, RoutingKey, []),
+    ?assertEqual(Dsts, lists:sort(Actual)),
+    ?assertEqual({[Dst1], #{}}, rabbit_db_topic_exchange:match(Src2, RoutingKey, [])),
     ?assertEqual(ok, rabbit_db_topic_exchange:delete_all_for_exchange(Src1)),
-    ?assertEqual([], rabbit_db_topic_exchange:match(Src1, RoutingKey)),
-    ?assertEqual([Dst1], rabbit_db_topic_exchange:match(Src2, RoutingKey)),
+    ?assertEqual({[], #{}}, rabbit_db_topic_exchange:match(Src1, RoutingKey, [])),
+    ?assertEqual({[Dst1], #{}}, rabbit_db_topic_exchange:match(Src2, RoutingKey, [])),
     passed.
 
-match(Config) ->
-    passed = rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, match1, [Config]).
+match_v1(Config) ->
+    passed = rabbit_ct_broker_helpers:rpc(Config, ?MODULE, match_v1_0, [Config]).
 
-match1(_Config) ->
+match_v1_0(_Config) ->
     Src = rabbit_misc:r(?VHOST, exchange, <<"test-exchange">>),
     Dst1 = rabbit_misc:r(?VHOST, queue, <<"test-queue1">>),
     Dst2 = rabbit_misc:r(?VHOST, queue, <<"test-queue2">>),
@@ -137,20 +140,123 @@ match1(_Config) ->
     Dst4 = rabbit_misc:r(?VHOST, queue, <<"test-queue4">>),
     Dst5 = rabbit_misc:r(?VHOST, queue, <<"test-queue5">>),
     Dst6 = rabbit_misc:r(?VHOST, queue, <<"test-queue6">>),
+    Dst7 = rabbit_misc:r(?VHOST, queue, <<"test-queue7">>),
     set(Src, <<"a.b.c">>, Dst1),
     set(Src, <<"a.*.c">>, Dst2),
     set(Src, <<"*.#">>, Dst3),
     set(Src, <<"#">>, Dst4),
     set(Src, <<"#.#">>, Dst5),
     set(Src, <<"a.*">>, Dst6),
-    Dsts1 = lists:sort([Dst1, Dst2, Dst3, Dst4, Dst5]),
-    ?assertEqual(Dsts1, lists:usort(rabbit_db_topic_exchange:match(Src, <<"a.b.c">>))),
-    Dsts2 = lists:sort([Dst3, Dst4, Dst5, Dst6]),
-    ?assertEqual(Dsts2, lists:usort(rabbit_db_topic_exchange:match(Src, <<"a.b">>))),
-    Dsts3 = lists:sort([Dst4, Dst5]),
-    ?assertEqual(Dsts3, lists:usort(rabbit_db_topic_exchange:match(Src, <<"">>))),
-    Dsts4 = lists:sort([Dst3, Dst4, Dst5]),
-    ?assertEqual(Dsts4, lists:usort(rabbit_db_topic_exchange:match(Src, <<"zen.rabbit">>))),
+    set(Src, <<"a.b.#">>, Dst7),
+    {Dsts1, Qs1} = rabbit_db_topic_exchange:match(Src, <<"a.b.c">>, []),
+    ?assertEqual(lists:sort([Dst1, Dst2, Dst3, Dst4, Dst5, Dst7]),
+                 lists:usort(Dsts1)),
+    {Dsts2, Qs2} = rabbit_db_topic_exchange:match(Src, <<"a.b">>, []),
+    ?assertEqual(lists:sort([Dst3, Dst4, Dst5, Dst6, Dst7]),
+                 lists:usort(Dsts2)),
+    {Dsts3, Qs3} = rabbit_db_topic_exchange:match(Src, <<"">>, []),
+    ?assertEqual(lists:sort([Dst4, Dst5]),
+                 lists:usort(Dsts3)),
+    {Dsts4, Qs4} = rabbit_db_topic_exchange:match(Src, <<"zen.rabbit">>, []),
+    ?assertEqual(lists:sort([Dst3, Dst4, Dst5]),
+                 lists:usort(Dsts4)),
+    ?assert(lists:all(fun(Qs) -> maps:size(Qs) =:= 0 end, [Qs1, Qs2, Qs3, Qs4])),
+    passed.
+
+match_v2_many_destinations(Config) ->
+    passed = rabbit_ct_broker_helpers:rpc(
+               Config, ?MODULE, match_v2_many_destinations0, []).
+
+match_v2_many_destinations0() ->
+    Src = rabbit_misc:r(?VHOST, exchange, <<"test-exchange">>),
+    Dst1 = rabbit_misc:r(?VHOST, queue, <<"mqtt-subscription-q1">>),
+    Dst2 = rabbit_misc:r(?VHOST, queue, <<"mqtt-subscription-q2">>),
+    Dst3 = rabbit_misc:r(?VHOST, queue, <<"mqtt-subscription-q3">>),
+    Dst4 = rabbit_misc:r(?VHOST, queue, <<"mqtt-subscription-q4">>),
+    Dst5 = rabbit_misc:r(?VHOST, queue, <<"mqtt-subscription-q5">>),
+    Dst6 = rabbit_misc:r(?VHOST, queue, <<"mqtt-subscription-q6">>),
+    Dst7 = rabbit_misc:r(?VHOST, queue, <<"mqtt-subscription-q7">>),
+    Dst8 = rabbit_misc:r(?VHOST, exchange, <<"e1">>),
+    Dst9 = rabbit_misc:r(?VHOST, exchange, <<"e2">>),
+    set(Src, <<"a.b.c">>, Dst1),
+    set(Src, <<"a.*.c">>, Dst2),
+    set(Src, <<"*.#">>, Dst3),
+    set(Src, <<"#">>, Dst4),
+    set(Src, <<"#.#">>, Dst5),
+    set(Src, <<"a.*">>, Dst6),
+    set(Src, <<"a.b.#">>, Dst7),
+    set(Src, <<"a.b.c">>, Dst8),
+    set(Src, <<"#">>, Dst9),
+
+    Mod = rabbit_feature_flags,
+    ok = meck:new(Mod),
+    ok = meck:expect(Mod, is_enabled, [mqtt_v5], true),
+    Opts = [v2],
+
+    {Exchanges1, Qs1} = rabbit_db_topic_exchange:match(Src, <<"a.b.c">>, Opts),
+    ?assertMatch(#{Dst1 := [<<"a.b.c">>],
+                   Dst2 := [<<"a.*.c">>],
+                   Dst3 := [<<"*.#">>],
+                   Dst4 := [<<"#">>],
+                   Dst5 := [<<"#.#">> | _],
+                   Dst7 := [<<"a.b.#">>]},
+                 Qs1),
+    ?assertEqual(6, map_size(Qs1)),
+    ?assertEqual(lists:sort([Dst8, Dst9]),
+                 lists:sort(Exchanges1)),
+
+    {Exchanges2, Qs2} = rabbit_db_topic_exchange:match(Src, <<"a.b">>, Opts),
+    ?assertMatch(#{Dst3 := [<<"*.#">>],
+                   Dst4 := [<<"#">>],
+                   Dst5 := [<<"#.#">> | _],
+                   Dst6 := [<<"a.*">>],
+                   Dst7 := [<<"a.b.#">>]},
+                 Qs2),
+    ?assertEqual(5, map_size(Qs2)),
+    ?assertEqual([Dst9], Exchanges2),
+
+    {Exchanges3, Qs3} = rabbit_db_topic_exchange:match(Src, <<"">>, Opts),
+    ?assertMatch(#{Dst4 := [<<"#">>],
+                   Dst5 := [<<"#.#">> | _]},
+                 Qs3),
+    ?assertEqual(2, map_size(Qs3)),
+    ?assertEqual([Dst9], Exchanges3),
+
+    {Exchanges4, Qs4} = rabbit_db_topic_exchange:match(Src, <<"zen.rabbit">>, Opts),
+    ?assertMatch(#{Dst3 := [<<"*.#">>],
+                   Dst4 := [<<"#">>],
+                   Dst5 := [<<"#.#">> | _]},
+                 Qs4),
+    ?assertEqual(3, map_size(Qs4)),
+    ?assertEqual([Dst9], Exchanges4),
+
+    ?assert(meck:validate(Mod)),
+    ok = meck:unload(Mod),
+    passed.
+
+match_v2_single_destination(Config) ->
+    passed = rabbit_ct_broker_helpers:rpc(
+               Config, ?MODULE, match_v2_single_destination0, []).
+
+match_v2_single_destination0() ->
+    Src = rabbit_misc:r(?VHOST, exchange, <<"test-exchange">>),
+    Dst = rabbit_misc:r(?VHOST, queue, <<"mqtt-subscription-myclientqos1">>),
+    [set(Src, BKey, Dst) ||
+     BKey <- [<<"a.b.c">>, <<"a.*.c">>, <<"*.#">>, <<"#">>, <<"#.#">>, <<"a.*">>, <<"a.b.#">>]],
+
+    Mod = rabbit_feature_flags,
+    ok = meck:new(Mod),
+    ok = meck:expect(Mod, is_enabled, [mqtt_v5], true),
+    Opts = [v2],
+
+    {[], Qs} = rabbit_db_topic_exchange:match(Src, <<"a.b.c">>, Opts),
+    ?assertEqual(1, map_size(Qs)),
+    BindingKeys = maps:get(Dst, Qs),
+    ?assertEqual(lists:sort([<<"a.b.c">>, <<"a.*.c">>, <<"*.#">>, <<"#">>, <<"#.#">>, <<"a.b.#">>]),
+                 lists:usort(BindingKeys)),
+
+    ?assert(meck:validate(Mod)),
+    ok = meck:unload(Mod),
     passed.
 
 set(Src, RoutingKey, Dst) ->
diff --git a/deps/rabbitmq_mqtt/BUILD.bazel b/deps/rabbitmq_mqtt/BUILD.bazel
index 5de3cf5238..6a8e2f1dfe 100644
--- a/deps/rabbitmq_mqtt/BUILD.bazel
+++ b/deps/rabbitmq_mqtt/BUILD.bazel
@@ -256,11 +256,11 @@ rabbitmq_suite(
 
 rabbitmq_integration_suite(
     name = "v5_SUITE",
+    size = "large",
     additional_beam = [
         ":test_util_beam",
     ],
-    # shard_count = 4,
-    shard_count = 2,
+    shard_count = 4,
     runtime_deps = [
         "@emqtt//:erlang_app",
         "@gun//:erlang_app",
diff --git a/deps/rabbitmq_mqtt/app.bzl b/deps/rabbitmq_mqtt/app.bzl
index e8aaa46d82..9c95326f9b 100644
--- a/deps/rabbitmq_mqtt/app.bzl
+++ b/deps/rabbitmq_mqtt/app.bzl
@@ -329,4 +329,5 @@ def test_suite_beam_files(name = "test_suite_beam_files"):
         outs = ["test/v5_SUITE.beam"],
         app_name = "rabbitmq_mqtt",
         erlc_opts = "//:test_erlc_opts",
+        deps = ["//deps/amqp_client:erlang_app"],
     )
diff --git a/deps/rabbitmq_mqtt/include/rabbit_mqtt_packet.hrl b/deps/rabbitmq_mqtt/include/rabbit_mqtt_packet.hrl
index b528c85779..28e87350a7 100644
--- a/deps/rabbitmq_mqtt/include/rabbit_mqtt_packet.hrl
+++ b/deps/rabbitmq_mqtt/include/rabbit_mqtt_packet.hrl
@@ -32,6 +32,9 @@
 -define(MAX_PACKET_SIZE, ?VARIABLE_BYTE_INTEGER_MAX).
 -type max_packet_size() :: 1..?MAX_PACKET_SIZE.
 
+%% "The Subscription Identifier can have the value of 1 to 268,435,455." [v5 3.8.2.1.2]
+-type subscription_identifier() :: 1..?VARIABLE_BYTE_INTEGER_MAX.
+
 -define(UINT_MAX, 16#FFFFFFFF).
 
 %% MQTT Control Packet types
@@ -185,12 +188,15 @@
                              props = #{} :: properties()
                             }).
 
+-record(mqtt_subscription_opts, {qos :: qos(), % maximum QoS
+                                 no_local = false :: boolean(),
+                                 retain_as_published = false :: boolean(),
+                                 retain_handling = 0 :: 0..2,
+                                 id :: option(subscription_identifier())
+                                }).
+
 -record(mqtt_subscription, {topic_filter :: binary(),
-                            %% subscription options
-                            qos :: qos(), % maximum QoS
-                            no_local :: boolean(),
-                            retain_as_published :: boolean(),
-                            retain_handling :: 0..2
+                            options :: #mqtt_subscription_opts{}
                            }).
 
 -record(mqtt_packet_subscribe, {packet_id :: packet_id(),
diff --git a/deps/rabbitmq_mqtt/src/rabbit_mqtt_ff.erl b/deps/rabbitmq_mqtt/src/rabbit_mqtt_ff.erl
index c7ca1cc9a6..4b7475b439 100644
--- a/deps/rabbitmq_mqtt/src/rabbit_mqtt_ff.erl
+++ b/deps/rabbitmq_mqtt/src/rabbit_mqtt_ff.erl
@@ -24,6 +24,14 @@
       callbacks     => #{enable => {mqtt_node, delete}}
      }}).
 
+%% This feature flag is needed to prevent clients from downgrading an MQTT v5 session to v3 or v4 on
+%% a lower version node. Such a session downgrade (or upgrade) requires changing binding arguments.
+-rabbit_feature_flag(
+   {mqtt_v5,
+    #{desc          => "Support MQTT 5.0",
+      stability     => stable
+     }}).
+
 -spec track_client_id_in_ra() -> boolean().
 track_client_id_in_ra() ->
     rabbit_feature_flags:is_disabled(delete_ra_cluster_mqtt_node).
diff --git a/deps/rabbitmq_mqtt/src/rabbit_mqtt_packet.erl b/deps/rabbitmq_mqtt/src/rabbit_mqtt_packet.erl
index 2d0d911fb5..8e07b1f824 100644
--- a/deps/rabbitmq_mqtt/src/rabbit_mqtt_packet.erl
+++ b/deps/rabbitmq_mqtt/src/rabbit_mqtt_packet.erl
@@ -131,12 +131,15 @@ parse_packet(Bin, #mqtt_packet_fixed{type = Type,
             %% "SUBSCRIBE messages use QoS level 1 to acknowledge multiple subscription requests."
             1 = Qos,
             {Props, Rest1} = parse_props(PacketBin, ProtoVer),
+            Id = maps:get('Subscription-Identifier', Props, undefined),
             Subscriptions = [#mqtt_subscription{
                                 topic_filter = Topic,
-                                qos = QoS,
-                                no_local = int_to_bool(Nl),
-                                retain_as_published = int_to_bool(Rap),
-                                retain_handling = Rh} ||
+                                options = #mqtt_subscription_opts{
+                                             qos = QoS,
+                                             no_local = int_to_bool(Nl),
+                                             retain_as_published = int_to_bool(Rap),
+                                             retain_handling = Rh,
+                                             id = Id}} ||
                              <<Len:16, Topic:Len/binary, _Reserved:2, Rh:2, Rap:1, Nl:1, QoS:2>> <= Rest1],
             Subscribe = #mqtt_packet_subscribe{packet_id = PacketId,
                                                props = Props,
@@ -311,6 +314,8 @@ parse_prop(<<16#09, Len:16, Val:Len/binary, Bin/binary>>, Props) ->
     parse_prop(Bin, Props#{'Correlation-Data' => Val});
 parse_prop(<<16#0B, Bin/binary>>, Props) ->
     {Val, Rest} = parse_variable_byte_integer(Bin),
+    %% Client sends at most one Subscription-Identifier to the server (in SUBSCRIBE packet).
+    %% "It is a Protocol Error to include the Subscription Identifier more than once." [v5 3.8.2.1.2]
     parse_prop(Rest, Props#{'Subscription-Identifier' => Val});
 parse_prop(<<16#11, Val:32, Bin/binary>>, Props) ->
     parse_prop(Bin, Props#{'Session-Expiry-Interval' => Val});
@@ -552,12 +557,13 @@ serialise_prop('Response-Topic', Val) ->
     [16#08, serialise_binary(Val)];
 serialise_prop('Correlation-Data', Val) ->
     [16#09, serialise_binary(Val)];
-serialise_prop('Subscription-Identifier', Val) ->
-    %%TODO Allow for multiple subscription identifiers
-    %% "Multiple Subscription Identifiers will be included if the publication is the result
-    %% of a match to more than one subscription, in this case their order is not significant."
-    %% [v5 3.3.2.3.8]
-    [16#0B, serialise_variable_byte_integer(Val)];
+serialise_prop('Subscription-Identifier', Ids)
+%% Server can send multiple Subscription-Identifiers to the client (in PUBLISH packet).
+%% "Multiple Subscription Identifiers will be included if the publication is the result
+%% of a match to more than one subscription, in this case their order is not significant."
+%% [v5 3.3.2.3.8]
+  when is_list(Ids) ->
+    [[16#0B, serialise_variable_byte_integer(Id)] || Id <- Ids];
 serialise_prop('Session-Expiry-Interval', Val) ->
     <<16#11, Val:32>>;
 serialise_prop('Assigned-Client-Identifier', Val) ->
diff --git a/deps/rabbitmq_mqtt/src/rabbit_mqtt_processor.erl b/deps/rabbitmq_mqtt/src/rabbit_mqtt_processor.erl
index c53029ebc9..8ab213eff3 100644
--- a/deps/rabbitmq_mqtt/src/rabbit_mqtt_processor.erl
+++ b/deps/rabbitmq_mqtt/src/rabbit_mqtt_processor.erl
@@ -41,6 +41,7 @@
 
 -type send_fun() :: fun((iodata()) -> ok).
 -type session_expiry_interval() :: non_neg_integer() | infinity.
+-type subscriptions() :: #{TopicFilter :: binary() => #mqtt_subscription_opts{}}.
 
 -record(auth_state,
         {user :: #user{},
@@ -87,7 +88,7 @@
          %% (Not to be confused with packet IDs sent from client to server which can be the
          %% same IDs because client and server assign IDs independently of each other.)
          packet_id = 1 :: packet_id(),
-         subscriptions = #{} :: #{TopicFilter :: binary() => QoS :: ?QOS_0..?QOS_1},
+         subscriptions = #{} :: subscriptions(),
          auth_state = #auth_state{},
          ra_register_state :: option(registered | {pending, reference()}),
          %% quorum queues and streams whose soft limit has been exceeded
@@ -222,7 +223,6 @@ process_connect(
             Props0 = #{'Maximum-QoS' => ?QOS_1,
                        'Maximum-Packet-Size' => persistent_term:get(
                                                   ?PERSISTENT_TERM_MAX_PACKET_SIZE_AUTHENTICATED),
-                       'Subscription-Identifier-Available' => 0,
                        'Shared-Subscription-Available' => 0,
                        'Session-Expiry-Interval' => case SessionExpiry of
                                                         infinity -> ?UINT_MAX;
@@ -297,7 +297,7 @@ process_connect(State0) ->
     maybe
         {ok, QoS0SessPresent, State1} ?= handle_clean_start_qos0(State0),
         {ok, SessPresent, State2} ?= handle_clean_start_qos1(QoS0SessPresent, State1),
-        State = cache_subscriptions(SessPresent, State2),
+        {ok, State} ?= init_subscriptions(SessPresent, State2),
         rabbit_networking:register_non_amqp_connection(self()),
         self() ! connection_created,
         {ok, SessPresent, State}
@@ -415,15 +415,17 @@ process_request(?SUBSCRIBE,
               %% to close the client connection anyway.
               {[E | L], S};
          (#mqtt_subscription{topic_filter = Topic,
-                             qos = TopicQos},
+                             options = Opts0},
           {L, S0}) ->
-              QoS = maybe_downgrade_qos(TopicQos),
+              QoS = maybe_downgrade_qos(Opts0#mqtt_subscription_opts.qos),
+              Opts = Opts0#mqtt_subscription_opts{qos = QoS},
               maybe
-                  ok ?= maybe_replace_old_sub(Topic, QoS, S0),
                   {ok, Q} ?= ensure_queue(QoS, S0),
                   QName = amqqueue:get_name(Q),
-                  ok ?= bind(QName, Topic, S0),
-                  Subs = maps:put(Topic, QoS, S0#state.subscriptions),
+                  BindingArgs = binding_args_for_proto_ver(ProtoVer, Opts),
+                  ok ?= add_subscription(Topic, BindingArgs, QName, S0),
+                  ok ?= maybe_delete_old_subscription(Topic, Opts, S0),
+                  Subs = maps:put(Topic, Opts, S0#state.subscriptions),
                   S1 = S0#state{subscriptions = Subs},
                   maybe_increment_consumer(S0, S1),
                   case self_consumes(Q) of
@@ -462,14 +464,16 @@ process_request(?UNSUBSCRIBE,
                                                                  topic_filters = Topics},
                              payload = undefined},
                 State0) ->
-    ?LOG_DEBUG("Received an UNSUBSCRIBE for topic(s) ~p", [Topics]),
+    ?LOG_DEBUG("Received an UNSUBSCRIBE for topic filter(s) ~p", [Topics]),
     {ReasonCodes, State} =
     lists:foldl(
-      fun(Topic, {L, #state{subscriptions = Subs0} = S0}) ->
+      fun(Topic, {L, #state{subscriptions = Subs0,
+                            cfg = #cfg{proto_ver = ProtoVer}} = S0}) ->
               case maps:take(Topic, Subs0) of
-                  {QoS, Subs} ->
-                      QName = queue_name(QoS, S0),
-                      case unbind(QName, Topic, S0) of
+                  {Opts, Subs} ->
+                      BindingArgs = binding_args_for_proto_ver(ProtoVer, Opts),
+                      case delete_subscription(
+                             Topic, BindingArgs, Opts#mqtt_subscription_opts.qos, S0) of
                           ok ->
                               S = S0#state{subscriptions = Subs},
                               maybe_decrement_consumer(S0, S),
@@ -579,12 +583,15 @@ update_session_expiry_interval(QName, Expiry) ->
             ok = rabbit_queue_type:policy_changed(Q) % respects queue args
     end.
 
-check_protocol_version(ProtoVersion) ->
-    case lists:member(ProtoVersion, proplists:get_keys(?PROTOCOL_NAMES)) of
+check_protocol_version(V)
+  when V =:= 3 orelse V =:= 4 ->
+    ok;
+check_protocol_version(5) ->
+    case rabbit_feature_flags:is_enabled(mqtt_v5) of
         true ->
             ok;
         false ->
-            ?LOG_ERROR("unsupported MQTT protocol version: ~p", [ProtoVersion]),
+            ?LOG_ERROR("Rejecting MQTT 5.0 connection because feature flag mqtt_v5 is disabled"),
             {error, ?RC_UNSUPPORTED_PROTOCOL_VERSION}
     end.
 
@@ -798,44 +805,84 @@ maybe_set_queue_qos1(?QOS_1, State = #state{cfg = Cfg = #cfg{queue_qos1 = undefi
 maybe_set_queue_qos1(_, State) ->
     State.
 
-%% Query subscriptions from the database and hold them in process state
-%% to avoid future mnesia:match_object/3 queries.
-cache_subscriptions(_SessionPresent = _SubscriptionsPresent = true,
-                    State = #state{cfg = #cfg{proto_ver = ProtoVer}}) ->
-    SubsQos0 = topic_names(?QOS_0, State),
-    SubsQos1 = topic_names(?QOS_1, State),
-    Subs = maps:merge(maps:from_keys(SubsQos0, ?QOS_0),
-                      maps:from_keys(SubsQos1, ?QOS_1)),
-    rabbit_global_counters:consumer_created(ProtoVer),
-    State#state{subscriptions = Subs};
-cache_subscriptions(_, State) ->
-    State.
+-spec init_subscriptions(boolean(), state()) ->
+    {ok, state()} | {error, reason_code()}.
+init_subscriptions(_SessionPresent = _SubscriptionsPresent = true,
+                   State = #state{cfg = #cfg{proto_ver = ProtoVer}}) ->
+    maybe
+        {ok, SubsQos0} ?= init_subscriptions0(?QOS_0, State),
+        {ok, SubsQos1} ?= init_subscriptions0(?QOS_1, State),
+        Subs = maps:merge(SubsQos0, SubsQos1),
+        rabbit_global_counters:consumer_created(ProtoVer),
+        %% Cache subscriptions in process state to avoid future mnesia:match_object/3 queries.
+        {ok, State#state{subscriptions = Subs}}
+    end;
+init_subscriptions(_, State) ->
+    {ok, State}.
 
-topic_names(QoS, State = #state{cfg = #cfg{exchange = Exchange}}) ->
+-spec init_subscriptions0(qos(), state()) ->
+    {ok, subscriptions()} | {error, reason_code()}.
+init_subscriptions0(QoS, State0 = #state{cfg = #cfg{proto_ver = ProtoVer,
+                                                    exchange = Exchange}}) ->
     Bindings =
     rabbit_binding:list_for_source_and_destination(
       Exchange,
-      queue_name(QoS, State),
+      queue_name(QoS, State0),
       %% Querying table rabbit_route is catastrophic for CPU usage.
       %% Querying table rabbit_reverse_route is acceptable because
       %% the source exchange is always the same in the MQTT plugin whereas
       %% the destination queue is different for each MQTT client and
       %% rabbit_reverse_route is sorted by destination queue.
       _Reverse = true),
-    lists:map(fun(B) -> amqp_to_mqtt(B#binding.key) end, Bindings).
-
-%% "If a Server receives a SUBSCRIBE Packet containing a Topic Filter that is identical
-%% to an existing Subscription’s Topic Filter then it MUST completely replace that
-%% existing Subscription with a new Subscription. The Topic Filter in the new Subscription
-%% will be identical to that in the previous Subscription, although its maximum QoS value
-%% could be different." [MQTT-3.8.4-3].
-maybe_replace_old_sub(TopicName, QoS, State = #state{subscriptions = Subs}) ->
-    case Subs of
-        #{TopicName := OldQoS} when OldQoS =/= QoS ->
-            QName = queue_name(OldQoS, State),
-            unbind(QName, TopicName, State);
-        _ ->
-            ok
+    try
+        Subs = lists:foldl(
+                 fun(#binding{key = Key,
+                              args = Args = [Opts0 = #mqtt_subscription_opts{}]},
+                     Acc) ->
+                         TopicFilter = amqp_to_mqtt(Key),
+                         Opts = case ProtoVer of
+                                    ?MQTT_PROTO_V5 ->
+                                        Opts0;
+                                    _ ->
+                                        %% session downgrade
+                                        ok = recreate_subscription(TopicFilter, Args, [], QoS, State0),
+                                        #mqtt_subscription_opts{qos = QoS}
+                                end,
+                         maps:put(TopicFilter, Opts, Acc);
+                    (#binding{key = Key,
+                              args = Args = []},
+                     Acc) ->
+                         Opts = #mqtt_subscription_opts{qos = QoS},
+                         TopicFilter = amqp_to_mqtt(Key),
+                         case ProtoVer of
+                             ?MQTT_PROTO_V5 ->
+                                 %% session upgrade
+                                 ok = recreate_subscription(TopicFilter, Args, [Opts], QoS, State0);
+                             _ ->
+                                 ok
+                         end,
+                         maps:put(TopicFilter, Opts, Acc)
+                 end, #{}, Bindings),
+        {ok, Subs}
+    catch throw:{error, Reason} ->
+              Rc = case Reason of
+                       access_refused -> ?RC_NOT_AUTHORIZED;
+                       _Other -> ?RC_IMPLEMENTATION_SPECIFIC_ERROR
+                   end,
+              {error, Rc}
+    end.
+
+recreate_subscription(TopicFilter, OldBindingArgs, NewBindingArgs, Qos, State) ->
+    case add_subscription(TopicFilter, NewBindingArgs, Qos, State) of
+        ok ->
+            case delete_subscription(TopicFilter, OldBindingArgs, Qos, State) of
+                ok ->
+                    ok;
+                {error, _} = Err ->
+                    throw(Err)
+            end;
+        {error, _} = Err ->
+            throw(Err)
     end.
 
 -spec hand_off_to_retainer(pid(), binary(), mqtt_msg()) -> ok.
@@ -847,9 +894,12 @@ hand_off_to_retainer(RetainerPid, Topic0, Msg = #mqtt_msg{payload = Payload}) ->
            rabbit_mqtt_retainer:retain(RetainerPid, Topic, Msg)
     end.
 
-maybe_send_retained_message(RPid, #mqtt_subscription{topic_filter = Topic0,
-                                                     qos = SubscribeQos},
-                            State0 = #state{packet_id = PacketId0}) ->
+maybe_send_retained_message(
+  RPid,
+  #mqtt_subscription{topic_filter = Topic0,
+                     options = #mqtt_subscription_opts{
+                                  qos = SubscribeQos}},
+  State0 = #state{packet_id = PacketId0}) ->
     Topic = amqp_to_mqtt(Topic0),
     case rabbit_mqtt_retainer:fetch(RPid, Topic) of
         undefined ->
@@ -1266,25 +1316,53 @@ consume(Q, QoS, #state{
             Err
     end.
 
-bind(QName, TopicName, State) ->
-    binding_action_with_checks(QName, TopicName, add, State).
-
-unbind(QName, TopicName, State) ->
-    binding_action_with_checks(QName, TopicName, remove, State).
+binding_args_for_proto_ver(?MQTT_PROTO_V5, SubOpts) ->
+    [SubOpts];
+binding_args_for_proto_ver(_, _) ->
+    [].
+
+add_subscription(TopicFilter, BindingArgs, Qos, State)
+  when is_integer(Qos) ->
+    add_subscription(TopicFilter, BindingArgs, queue_name(Qos, State), State);
+add_subscription(TopicFilter, BindingArgs, QName, State) ->
+    binding_action_with_checks(QName, TopicFilter, BindingArgs, add, State).
+
+delete_subscription(TopicFilter, BindingArgs, Qos, State) ->
+    binding_action_with_checks(
+      queue_name(Qos, State), TopicFilter, BindingArgs, remove, State).
+
+%% "If a Server receives a SUBSCRIBE packet containing a Topic Filter that is identical to a
+%% Non‑shared Subscription’s Topic Filter for the current Session, then it MUST replace that
+%% existing Subscription with a new Subscription [MQTT-3.8.4-3]. The Topic Filter in the new
+%% Subscription will be identical to that in the previous Subscription, although its
+%% Subscription Options could be different." [v5 3.8.4]
+maybe_delete_old_subscription(TopicFilter, Opts, State = #state{subscriptions = Subs,
+                                                                cfg = #cfg{proto_ver = ProtoVer}}) ->
+    case Subs of
+        #{TopicFilter := OldOpts}
+          when OldOpts =/= Opts ->
+            delete_subscription(TopicFilter,
+                                binding_args_for_proto_ver(ProtoVer, OldOpts),
+                                OldOpts#mqtt_subscription_opts.qos,
+                                State);
+        _ ->
+            ok
+    end.
 
-binding_action_with_checks(QName, TopicName, Action,
+binding_action_with_checks(QName, TopicFilter, BindingArgs, Action,
                            State = #state{cfg = #cfg{exchange = ExchangeName},
                                           auth_state = AuthState}) ->
     %% Same permissions required for binding or unbinding queue to/from topic exchange.
     maybe
         ok ?= check_queue_write_access(QName, AuthState),
         ok ?= check_exchange_read_access(ExchangeName, AuthState),
-        ok ?= check_topic_access(TopicName, read, State),
-        ok ?= binding_action(ExchangeName, TopicName, QName, fun rabbit_binding:Action/2, AuthState)
+        ok ?= check_topic_access(TopicFilter, read, State),
+        ok ?= binding_action(ExchangeName, TopicFilter, QName, BindingArgs,
+                             fun rabbit_binding:Action/2, AuthState)
     else
         {error, Reason} = Err ->
-            ?LOG_ERROR("Failed to ~s binding between ~s and ~s for topic ~s: ~p",
-                       [Action, rabbit_misc:rs(ExchangeName), rabbit_misc:rs(QName), TopicName, Reason]),
+            ?LOG_ERROR("Failed to ~s binding between ~s and ~s for topic filter ~s: ~p",
+                       [Action, rabbit_misc:rs(ExchangeName), rabbit_misc:rs(QName), TopicFilter, Reason]),
             Err
     end.
 
@@ -1298,15 +1376,18 @@ check_exchange_read_access(ExchangeName, #auth_state{user = User,
     %% read access to exchange required for queue.(un)bind
     check_resource_access(User, ExchangeName, read, AuthzCtx).
 
-binding_action(ExchangeName, TopicName, QName, BindingFun, #auth_state{user = #user{username = Username}}) ->
-    RoutingKey = mqtt_to_amqp(TopicName),
+binding_action(ExchangeName, TopicFilter, QName, BindingArgs,
+               BindingFun, #auth_state{user = #user{username = Username}}) ->
+    RoutingKey = mqtt_to_amqp(TopicFilter),
     Binding = #binding{source = ExchangeName,
                        destination = QName,
-                       key = RoutingKey},
+                       key = RoutingKey,
+                       args = BindingArgs},
     BindingFun(Binding, Username).
 
 publish_to_queues(
-  #mqtt_msg{qos = Qos,
+  #mqtt_msg{retain = Retain,
+            qos = Qos,
             topic = Topic,
             packet_id = PacketId,
             payload = Payload,
@@ -1327,7 +1408,8 @@ publish_to_queues(
                                       {undefined, undefined}
                               end,
     PBasic = #'P_basic'{
-                headers = [{<<"x-mqtt-publish-qos">>, byte, Qos}],
+                headers = [{<<"x-mqtt-publish-qos">>, byte, Qos},
+                           {<<"x-mqtt-retain">>, bool, Retain}],
                 delivery_mode = delivery_mode(Qos),
                 expiration = Expiration,
                 timestamp = Timestamp},
@@ -1357,7 +1439,8 @@ publish_to_queues(
                  },
     case rabbit_exchange:lookup(ExchangeName) of
         {ok, Exchange} ->
-            QNames = rabbit_exchange:route(Exchange, Delivery),
+            QNames0 = rabbit_exchange:route(Exchange, Delivery, [v2]),
+            QNames = drop_local(QNames0, State),
             rabbit_trace:tap_in(BasicMessage, QNames, ConnName, Username, TraceState),
             deliver_to_queues(Delivery, QNames, State);
         {error, not_found} ->
@@ -1365,6 +1448,39 @@ publish_to_queues(
             {error, exchange_not_found, State}
     end.
 
+%% "Bit 2 of the Subscription Options represents the No Local option.
+%% If the value is 1, Application Messages MUST NOT be forwarded to a connection with a ClientID
+%% equal to the ClientID of the publishing connection [MQTT-3.8.3-3]." [v5 3.8.3.1]
+drop_local(QNames, #state{subscriptions = Subs,
+                          cfg = #cfg{proto_ver = ?MQTT_PROTO_V5,
+                                     vhost = Vhost,
+                                     client_id = ClientId}}) ->
+    ClientIdSize = byte_size(ClientId),
+    lists:filter(
+      fun({#resource{virtual_host = Vhost0,
+                     name = <<"mqtt-subscription-",
+                              ClientId0:ClientIdSize/binary,
+                              "qos", _:1/binary >>},
+           BindingKeys})
+            when Vhost0 =:= Vhost andalso
+                 ClientId0 =:= ClientId ->
+              lists:any(
+                fun(BKey) ->
+                        TopicFilter = amqp_to_mqtt(BKey),
+                        case Subs of
+                            #{TopicFilter := #mqtt_subscription_opts{
+                                                no_local = NoLocal}} ->
+                                not NoLocal;
+                            _ ->
+                                true
+                        end
+                end, BindingKeys);
+         (_) ->
+              true
+      end, QNames);
+drop_local(QNames, _) ->
+    QNames.
+
 deliver_to_queues(Delivery,
                   RoutedToQNames,
                   State0 = #state{queue_states = QStates0,
@@ -1399,9 +1515,10 @@ process_routing_confirm(#delivery{confirm = true,
                                   msg_seq_no = PktId},
                         [], State = #state{cfg = #cfg{proto_ver = ProtoVer}}) ->
     rabbit_global_counters:messages_unroutable_returned(ProtoVer, 1),
-    %% MQTT 5 spec:
+    %% TODO MQTT 5 spec:
     %% If the Server knows that there are no matching subscribers, it MAY use
     %% Reason Code 0x10 (No matching subscribers) instead of 0x00 (Success).
+    %% Also return "No matching subscribers" if message dropped due to NoLocal flag?
     send_puback(PktId, State),
     State;
 process_routing_confirm(#delivery{confirm = false}, _, State) ->
@@ -1413,7 +1530,7 @@ process_routing_confirm(#delivery{confirm = true,
 process_routing_confirm(#delivery{confirm = true,
                                   msg_seq_no = PktId},
                         Qs, State = #state{unacked_client_pubs = U0}) ->
-    QNames = lists:map(fun amqqueue:get_name/1, Qs),
+    QNames = rabbit_amqqueue:queue_names(Qs),
     U = rabbit_mqtt_confirms:insert(PktId, QNames, U0),
     State#state{unacked_client_pubs = U}.
 
@@ -1729,16 +1846,19 @@ maybe_publish_to_client(
          #basic_message{
             routing_keys = [RoutingKey | _CcRoutes],
             content = #content{payload_fragments_rev = FragmentsRev,
-                               properties = PBasic}}},
+                               properties = PBasic = #'P_basic'{headers = Headers}}}},
   QoS, State0) ->
-    Props = p_basic_to_publish_properties(PBasic),
+    MatchedTopicFilters = matched_topic_filters_v5(Headers, State0),
+    Props0 = p_basic_to_publish_properties(PBasic),
+    Props = maybe_add_subscription_ids(MatchedTopicFilters, Props0, State0),
     {PacketId, State} = msg_id_to_packet_id(QMsgId, QoS, State0),
     Packet =
     #mqtt_packet{
        fixed = #mqtt_packet_fixed{
                   type = ?PUBLISH,
                   qos = QoS,
-                  dup = Redelivered},
+                  dup = Redelivered,
+                  retain = retain(Headers, MatchedTopicFilters, State)},
        variable = #mqtt_packet_publish{
                      packet_id = PacketId,
                      topic_name = amqp_to_mqtt(RoutingKey),
@@ -1780,6 +1900,46 @@ p_basic_to_publish_properties(#'P_basic'{headers = Headers,
 p_basic_to_publish_properties(#'P_basic'{}) ->
     #{}.
 
+matched_topic_filters_v5(Headers, #state{cfg = #cfg{proto_ver = ?MQTT_PROTO_V5}}) ->
+    case rabbit_mqtt_util:table_lookup(Headers, <<"x-binding-keys">>) of
+        {array, BindingKeys} ->
+            [amqp_to_mqtt(BKey) || {longstr, BKey} <- BindingKeys];
+        undefined ->
+            []
+    end;
+matched_topic_filters_v5(_, _) ->
+    [].
+
+maybe_add_subscription_ids(TopicFilters, Props, #state{subscriptions = Subs}) ->
+    Ids = lists:filtermap(fun(T) -> case maps:get(T, Subs, undefined) of
+                                        #mqtt_subscription_opts{id = Id}
+                                          when is_integer(Id) ->
+                                            {true, Id};
+                                        _ ->
+                                            false
+                                    end
+                          end, TopicFilters),
+    case Ids of
+        [] -> Props;
+        _ -> maps:put('Subscription-Identifier', Ids, Props)
+    end.
+
+%% "Bit 3 of the Subscription Options represents the Retain As Published option.
+%% If 1, Application Messages forwarded using this subscription keep the RETAIN
+%% flag they were published with. If 0, Application Messages forwarded using
+%% this subscription have the RETAIN flag set to 0." [v5 3.8.3.1]
+retain(Headers, TopicFilters, #state{subscriptions = Subs}) ->
+    case rabbit_mqtt_util:table_lookup(Headers, <<"x-mqtt-retain">>) of
+        {bool, true} ->
+            lists:any(fun(T) -> case maps:get(T, Subs, undefined) of
+                                    #mqtt_subscription_opts{retain_as_published = Rap} -> Rap;
+                                    undefined -> false
+                                end
+                      end, TopicFilters);
+        _ ->
+            false
+    end.
+
 msg_id_to_packet_id(_, ?QOS_0, State) ->
     %% "A PUBLISH packet MUST NOT contain a Packet Identifier if its QoS value is set to 0 [MQTT-2.2.1-2]."
     {undefined, State};
@@ -1861,10 +2021,10 @@ publish_to_queues_with_checks(
                             Error
                     end;
                 {error, access_refused} ->
-                    {error, unauthorized, State}
+                    {error, access_refused, State}
             end;
         {error, access_refused} ->
-            {error, unauthorized, State}
+            {error, access_refused, State}
     end.
 
 check_resource_access(User, Resource, Perm, Context) ->
diff --git a/deps/rabbitmq_mqtt/src/rabbit_mqtt_reader.erl b/deps/rabbitmq_mqtt/src/rabbit_mqtt_reader.erl
index a55de5b587..9c48dda143 100644
--- a/deps/rabbitmq_mqtt/src/rabbit_mqtt_reader.erl
+++ b/deps/rabbitmq_mqtt/src/rabbit_mqtt_reader.erl
@@ -358,7 +358,7 @@ process_received_bytes(Bytes, State = #state{socket = Socket,
                               Rest,
                               State #state{parse_state = ParseState1,
                                            proc_state = ProcState1});
-                        {error, unauthorized = Reason, ProcState1} ->
+                        {error, access_refused = Reason, ProcState1} ->
                             ?LOG_ERROR("MQTT connection ~ts is closing due to an authorization failure", [ConnName]),
                             {stop, {shutdown, Reason}, pstate(State, ProcState1)};
                         {error, Reason, ProcState1} ->
diff --git a/deps/rabbitmq_mqtt/test/auth_SUITE.erl b/deps/rabbitmq_mqtt/test/auth_SUITE.erl
index d1999f909f..b1cc00b3bd 100644
--- a/deps/rabbitmq_mqtt/test/auth_SUITE.erl
+++ b/deps/rabbitmq_mqtt/test/auth_SUITE.erl
@@ -20,7 +20,9 @@
 
 -define(FAIL_IF_CRASH_LOG, {["Generic server.*terminating"],
                             fun () -> ct:fail(crash_detected) end}).
--import(rabbit_ct_broker_helpers, [rpc/5]).
+-import(rabbit_ct_broker_helpers,
+        [rpc/5,
+         set_full_permissions/3]).
 -import(util, [non_clean_sess_opts/0]).
 
 all() ->
@@ -126,7 +128,12 @@ init_per_group(authz, Config0) ->
     rabbit_ct_broker_helpers:add_user(Config1, User, Password),
     rabbit_ct_broker_helpers:add_vhost(Config1, VHost),
     [Log|_] = rpc(Config1, 0, rabbit, log_locations, []),
-    [{mqtt_user, User}, {mqtt_vhost, VHost}, {mqtt_password, Password}, {log_location, Log}|Config1];
+    Config2 = [{mqtt_user, User},
+               {mqtt_vhost, VHost},
+               {mqtt_password, Password},
+               {log_location, Log} | Config1],
+    ok = rabbit_ct_broker_helpers:enable_feature_flag(Config2, mqtt_v5),
+    Config2;
 init_per_group(Group, Config) ->
     Suffix = rabbit_ct_helpers:testcase_absname(Config, "", "-"),
     Config1 = rabbit_ct_helpers:set_config(Config, [
@@ -135,19 +142,22 @@ init_per_group(Group, Config) ->
     ]),
     MqttConfig = mqtt_config(Group),
     AuthConfig = auth_config(Group),
-    rabbit_ct_helpers:run_setup_steps(Config1,
-        [ fun(Conf) -> case MqttConfig of
-                           undefined  -> Conf;
-                           _          -> merge_app_env(MqttConfig, Conf)
-                       end
-          end] ++
-        [ fun(Conf) -> case AuthConfig of
-                            undefined -> Conf;
-                            _         -> merge_app_env(AuthConfig, Conf)
-                       end
-          end ] ++
-        rabbit_ct_broker_helpers:setup_steps() ++
-        rabbit_ct_client_helpers:setup_steps()).
+    Config2 = rabbit_ct_helpers:run_setup_steps(
+                Config1,
+                [fun(Conf) -> case MqttConfig of
+                                  undefined  -> Conf;
+                                  _          -> merge_app_env(MqttConfig, Conf)
+                              end
+                 end] ++
+                [fun(Conf) -> case AuthConfig of
+                                  undefined -> Conf;
+                                  _         -> merge_app_env(AuthConfig, Conf)
+                              end
+                 end] ++
+                rabbit_ct_broker_helpers:setup_steps() ++
+                rabbit_ct_client_helpers:setup_steps()),
+    ok = rabbit_ct_broker_helpers:enable_feature_flag(Config2, mqtt_v5),
+    Config2.
 
 end_per_group(G, Config)
   when G =:= v4;
@@ -208,7 +218,7 @@ init_per_testcase(user_credentials_auth, Config) ->
     User = <<"new-user">>,
     Pass = <<"new-user-pass">>,
     ok = rabbit_ct_broker_helpers:add_user(Config, 0, User, Pass),
-    ok = rabbit_ct_broker_helpers:set_full_permissions(Config, User, <<"/">>),
+    ok = set_full_permissions(Config, User, <<"/">>),
     Config1 = rabbit_ct_helpers:set_config(Config, [{new_user, User},
                                                     {new_user_pass, Pass}]),
     rabbit_ct_helpers:testcase_started(Config1, user_credentials_auth);
@@ -275,7 +285,7 @@ set_cert_user_on_default_vhost(Config) ->
     UserBin = rpc(Config, 0, rabbit_ssl, peer_cert_auth_name, [Cert]),
     User = binary_to_list(UserBin),
     ok = rabbit_ct_broker_helpers:add_user(Config, 0, User, ""),
-    ok = rabbit_ct_broker_helpers:set_full_permissions(Config, User, <<"/">>),
+    ok = set_full_permissions(Config, User, <<"/">>),
     rabbit_ct_helpers:set_config(Config, [{temp_ssl_user, User}]).
 
 set_vhost_for_cert_user(Config, User) ->
@@ -285,7 +295,7 @@ set_vhost_for_cert_user(Config, User) ->
         {<<"O=client,CN=unlikelytoexistuser">>, <<"vhost2">>}
     ],
     ok = rabbit_ct_broker_helpers:add_vhost(Config, VhostForCertUser),
-    ok = rabbit_ct_broker_helpers:set_full_permissions(Config, User, VhostForCertUser),
+    ok = set_full_permissions(Config, User, VhostForCertUser),
     ok = rabbit_ct_broker_helpers:set_global_parameter(Config, mqtt_default_vhosts, UserToVHostMappingParameter),
     rabbit_ct_helpers:set_config(Config, [{temp_vhost_for_ssl_user, VhostForCertUser}]).
 
@@ -301,7 +311,7 @@ set_vhost_for_port_vhost_mapping_user(Config, User) ->
 
     ],
     ok = rabbit_ct_broker_helpers:add_vhost(Config, VhostForPortMapping),
-    ok = rabbit_ct_broker_helpers:set_full_permissions(Config, User, VhostForPortMapping),
+    ok = set_full_permissions(Config, User, VhostForPortMapping),
     ok = rabbit_ct_broker_helpers:set_global_parameter(Config, mqtt_port_to_vhost_mapping, PortToVHostMappingParameter),
     rabbit_ct_helpers:set_config(Config, [{temp_vhost_for_port_mapping, VhostForPortMapping}]).
 
@@ -329,14 +339,14 @@ end_per_testcase(Testcase, Config) when Testcase == port_vhost_mapping_success;
                                         Testcase == port_vhost_mapping_not_allowed;
                                         Testcase == port_vhost_mapping_success_no_mapping ->
     User = <<"guest">>,
-    rabbit_ct_broker_helpers:set_full_permissions(Config, User, <<"/">>),
+    ok = set_full_permissions(Config, User, <<"/">>),
     VHost = ?config(temp_vhost_for_port_mapping, Config),
     ok = rabbit_ct_broker_helpers:delete_vhost(Config, VHost),
     ok = rabbit_ct_broker_helpers:clear_global_parameter(Config, mqtt_port_to_vhost_mapping),
     rabbit_ct_helpers:testcase_finished(Config, Testcase);
 end_per_testcase(port_vhost_mapping_vhost_does_not_exist, Config) ->
     User = <<"guest">>,
-    ok = rabbit_ct_broker_helpers:set_full_permissions(Config, User, <<"/">>),
+    ok = set_full_permissions(Config, User, <<"/">>),
     ok = rabbit_ct_broker_helpers:clear_global_parameter(Config, mqtt_port_to_vhost_mapping),
     rabbit_ct_helpers:testcase_finished(Config, port_vhost_mapping_vhost_does_not_exist);
 end_per_testcase(ssl_user_port_vhost_mapping_takes_precedence_over_cert_vhost_mapping, Config) ->
@@ -554,14 +564,14 @@ no_queue_bind_permission(Config) ->
      "'mqtt-subscription-mqtt-userqos0' in vhost 'mqtt-vhost' "
      "refused for user 'mqtt-user'",
      "Failed to add binding between exchange 'amq.topic' in vhost 'mqtt-vhost' and queue "
-     "'mqtt-subscription-mqtt-userqos0' in vhost 'mqtt-vhost' for topic test/topic: access_refused"
+     "'mqtt-subscription-mqtt-userqos0' in vhost 'mqtt-vhost' for topic filter test/topic: access_refused"
     ],
     test_subscribe_permissions_combination(<<".*">>, <<"">>, <<".*">>, Config, ExpectedLogs).
 
 no_queue_unbind_permission(Config) ->
     User = ?config(mqtt_user, Config),
     Vhost = ?config(mqtt_vhost, Config),
-    rabbit_ct_broker_helpers:set_permissions(Config, User, Vhost, <<".*">>, <<".*">>, <<".*">>),
+    set_full_permissions(Config, User, Vhost),
     P = rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_mqtt),
     Opts = [{host, "localhost"},
             {port, P},
@@ -576,27 +586,28 @@ no_queue_unbind_permission(Config) ->
                  emqtt:subscribe(C1, Topic, qos1)),
     ok = emqtt:disconnect(C1),
 
-    %% Revoke read access to amq.topic exchange.
-    rabbit_ct_broker_helpers:set_permissions(Config, User, Vhost, <<".*">>, <<".*">>, <<"^(?!amq\.topic$)">>),
+    %% Revoke write access to qos1 queue.
+    rabbit_ct_broker_helpers:set_permissions(Config, User, Vhost, <<".*">>, <<"mqtt-subscription-mqtt-userqos0">>, <<".*">>),
     {ok, C2} = emqtt:start_link(non_clean_sess_opts() ++ Opts),
     {ok, _} = emqtt:connect(C2),
     process_flag(trap_exit, true),
     %% We subscribe with the same client ID to the same topic again, but this time with QoS 0.
     %% Therefore we trigger the qos1 queue to be unbound (and the qos0 queue to be bound).
-    %% However, unbinding requires read access to the exchange, which we don't have anymore.
+    %% However, unbinding requires write access to the qos1 queue, which we don't have anymore.
     ExpectedReasonCode = suback_error_code(?RC_NOT_AUTHORIZED, Config),
     ?assertMatch({ok, _Properties, [ExpectedReasonCode]},
                  emqtt:subscribe(C2, Topic, qos0)),
     ok = assert_connection_closed(C2),
     ExpectedLogs =
-    ["MQTT resource access refused: read access to exchange 'amq.topic' in vhost 'mqtt-vhost' refused for user 'mqtt-user'",
+    ["MQTT resource access refused: write access to queue 'mqtt-subscription-mqtt-userqos1' "
+     "in vhost 'mqtt-vhost' refused for user 'mqtt-user'",
      "Failed to remove binding between exchange 'amq.topic' in vhost 'mqtt-vhost' and queue "
-     "'mqtt-subscription-mqtt-userqos1' in vhost 'mqtt-vhost' for topic my/topic: access_refused"
+     "'mqtt-subscription-mqtt-userqos1' in vhost 'mqtt-vhost' for topic filter my/topic: access_refused"
     ],
     wait_log(Config, [?FAIL_IF_CRASH_LOG, {ExpectedLogs, fun () -> stop end}]),
 
     %% Clean up the qos1 queue by connecting with clean session.
-    rabbit_ct_broker_helpers:set_permissions(Config, User, Vhost, <<".*">>, <<".*">>, <<".*">>),
+    set_full_permissions(Config, User, Vhost),
     {ok, C3} = emqtt:start_link([{clean_start, true} | Opts]),
     {ok, _} = emqtt:connect(C3),
     ok = emqtt:disconnect(C3).
@@ -775,7 +786,7 @@ no_topic_read_permission(Config) ->
               {["MQTT topic access refused: read access to topic 'test.topic' in exchange "
                 "'amq.topic' in vhost 'mqtt-vhost' refused for user 'mqtt-user'",
                 "Failed to add binding between exchange 'amq.topic' in vhost 'mqtt-vhost' and queue "
-                "'mqtt-subscription-mqtt-userqos0' in vhost 'mqtt-vhost' for topic test/topic: access_refused"
+                "'mqtt-subscription-mqtt-userqos0' in vhost 'mqtt-vhost' for topic filter test/topic: access_refused"
                ],
                fun () -> stop end}
              ]),
diff --git a/deps/rabbitmq_mqtt/test/command_SUITE.erl b/deps/rabbitmq_mqtt/test/command_SUITE.erl
index 85db3abdc7..e38c75293b 100644
--- a/deps/rabbitmq_mqtt/test/command_SUITE.erl
+++ b/deps/rabbitmq_mqtt/test/command_SUITE.erl
@@ -11,26 +11,22 @@
 -include_lib("eunit/include/eunit.hrl").
 -include_lib("amqp_client/include/amqp_client.hrl").
 -include("rabbit_mqtt.hrl").
--import(util, [connect/3]).
+-import(util, [connect/3, connect/4]).
 
 -define(COMMAND, 'Elixir.RabbitMQ.CLI.Ctl.Commands.ListMqttConnectionsCommand').
 
 all() ->
     [
+     {group, unit},
      {group, v4},
      {group, v5}
     ].
 
 groups() ->
     [
-     {v4, [shuffle], tests()},
-     {v5, [shuffle], tests()}
-    ].
-
-tests() ->
-    [
-     merge_defaults,
-     run
+     {unit, [], [merge_defaults]},
+     {v4, [], [run]},
+     {v5, [], [run]}
     ].
 
 suite() ->
@@ -56,8 +52,11 @@ end_per_suite(Config) ->
       rabbit_ct_client_helpers:teardown_steps() ++
       rabbit_ct_broker_helpers:teardown_steps()).
 
+init_per_group(unit, Config) ->
+    Config;
 init_per_group(Group, Config) ->
-    rabbit_ct_helpers:set_config(Config, {mqtt_version, Group}).
+    Config1 = rabbit_ct_helpers:set_config(Config, {mqtt_version, Group}),
+    util:maybe_skip_v5(Config1).
 
 end_per_group(_, Config) ->
     Config.
@@ -129,8 +128,18 @@ run(Config) ->
     ?assertEqual(InfoItemsSorted, lists:sort(proplists:get_keys(AllInfos1Con1))),
     ?assertEqual(InfoItemsSorted, lists:sort(proplists:get_keys(AllInfos2Con1))),
 
+    %% CLI command should list MQTT connections from all nodes.
+    C3 = connect(<<"simpleClient2">>, Config, 1, [{ack_timeout, 1}]),
+    rabbit_ct_helpers:eventually(
+      ?_assertEqual(
+         [[{client_id, <<"simpleClient">>}],
+          [{client_id, <<"simpleClient1">>}],
+          [{client_id, <<"simpleClient2">>}]],
+         lists:sort('Elixir.Enum':to_list(?COMMAND:run([<<"client_id">>], Opts))))),
+
     ok = emqtt:disconnect(C1),
-    ok = emqtt:disconnect(C2).
+    ok = emqtt:disconnect(C2),
+    ok = emqtt:disconnect(C3).
 
 start_amqp_connection(Type, Node, Port) ->
     amqp_connection:start(amqp_params(Type, Node, Port)).
diff --git a/deps/rabbitmq_mqtt/test/ff_SUITE.erl b/deps/rabbitmq_mqtt/test/ff_SUITE.erl
index 9e30914728..43c8ac9a43 100644
--- a/deps/rabbitmq_mqtt/test/ff_SUITE.erl
+++ b/deps/rabbitmq_mqtt/test/ff_SUITE.erl
@@ -27,8 +27,9 @@ all() ->
 
 groups() ->
     [
-     {cluster_size_3, [], [delete_ra_cluster_mqtt_node,
-                           rabbit_mqtt_qos0_queue]}
+     {cluster_size_3, [shuffle], [delete_ra_cluster_mqtt_node,
+                                  rabbit_mqtt_qos0_queue,
+                                  mqtt_v5]}
     ].
 
 suite() ->
@@ -60,10 +61,10 @@ end_per_group(_Group, Config) ->
 init_per_testcase(TestCase, Config) ->
     case rabbit_ct_broker_helpers:is_feature_flag_supported(Config, TestCase) of
         true ->
+            ?assertNot(rabbit_ct_broker_helpers:is_feature_flag_enabled(Config, TestCase)),
             Config;
         false ->
-            {skip, io_lib:format("feature flag ~s is unsupported",
-                                 [TestCase])}
+            {skip, io_lib:format("feature flag ~s is unsupported", [TestCase])}
     end.
 
 end_per_testcase(_TestCase, Config) ->
@@ -130,3 +131,15 @@ rabbit_mqtt_qos0_queue(Config) ->
                    messages_delivered_consume_auto_ack_total := 1},
                  get_global_counters(Config, ?PROTO_VER, 0, [{queue_type, FeatureFlag}])),
     ok = emqtt:disconnect(C2).
+
+mqtt_v5(Config) ->
+    FeatureFlag = ?FUNCTION_NAME,
+    {Client1, Connect} = util:start_client(?FUNCTION_NAME, Config, 0, [{proto_ver, v5}]),
+    unlink(Client1),
+    ?assertEqual({error, {unsupported_protocol_version, #{}}}, Connect(Client1)),
+
+    ?assertEqual(ok, rabbit_ct_broker_helpers:enable_feature_flag(Config, FeatureFlag)),
+
+    {Client2, Connect} = util:start_client(?FUNCTION_NAME, Config, 0, [{proto_ver, v5}]),
+    ?assertMatch({ok, _}, Connect(Client2)),
+    ok = emqtt:disconnect(Client2).
diff --git a/deps/rabbitmq_mqtt/test/packet_prop_SUITE.erl b/deps/rabbitmq_mqtt/test/packet_prop_SUITE.erl
index 33af737fc9..d22da9a8c1 100644
--- a/deps/rabbitmq_mqtt/test/packet_prop_SUITE.erl
+++ b/deps/rabbitmq_mqtt/test/packet_prop_SUITE.erl
@@ -138,7 +138,6 @@ publish_properties() ->
                         {'Response-Topic', utf8_string()},
                         {'Correlation-Data', binary_data()},
                         user_property(),
-                        {'Subscription-Identifier', non_zero_variable_byte_integer()},
                         {'Content-Type', utf8_string()}])),
          maps:from_list(L)).
 
diff --git a/deps/rabbitmq_mqtt/test/reader_SUITE.erl b/deps/rabbitmq_mqtt/test/reader_SUITE.erl
index daa7b22994..e65b5ec2af 100644
--- a/deps/rabbitmq_mqtt/test/reader_SUITE.erl
+++ b/deps/rabbitmq_mqtt/test/reader_SUITE.erl
@@ -67,10 +67,13 @@ init_per_suite(Config) ->
         {rmq_extra_tcp_ports, [tcp_port_mqtt_extra,
                                tcp_port_mqtt_tls_extra]}
       ]),
-    rabbit_ct_helpers:run_setup_steps(Config1,
-      [ fun merge_app_env/1 ] ++
-      rabbit_ct_broker_helpers:setup_steps() ++
-      rabbit_ct_client_helpers:setup_steps()).
+    Config2 = rabbit_ct_helpers:run_setup_steps(
+                Config1,
+                [fun merge_app_env/1] ++
+                rabbit_ct_broker_helpers:setup_steps() ++
+                rabbit_ct_client_helpers:setup_steps()),
+    ok = rabbit_ct_broker_helpers:enable_feature_flag(Config2, mqtt_v5),
+    Config2.
 
 end_per_suite(Config) ->
     rabbit_ct_helpers:run_teardown_steps(Config,
diff --git a/deps/rabbitmq_mqtt/test/retainer_SUITE.erl b/deps/rabbitmq_mqtt/test/retainer_SUITE.erl
index 3cd81ae62f..af278b86e9 100644
--- a/deps/rabbitmq_mqtt/test/retainer_SUITE.erl
+++ b/deps/rabbitmq_mqtt/test/retainer_SUITE.erl
@@ -44,7 +44,7 @@ tests() ->
     ].
 
 suite() ->
-    [{timetrap, {minutes, 1}}].
+    [{timetrap, {minutes, 2}}].
 
 %% -------------------------------------------------------------------
 %% Testsuite setup/teardown.
@@ -63,7 +63,7 @@ init_per_group(G, Config)
     rabbit_ct_helpers:set_config(Config, {mqtt_version, G});
 init_per_group(Group, Config0) ->
     Suffix = rabbit_ct_helpers:testcase_absname(Config0, "", "-"),
-    Config = rabbit_ct_helpers:set_config(
+    Config1 = rabbit_ct_helpers:set_config(
                Config0,
                [
                 {rmq_nodename_suffix, Suffix},
@@ -78,11 +78,13 @@ init_per_group(Group, Config0) ->
                      {default_vhost, "/"},
                      {default_permissions, [".*", ".*", ".*"]}
                     ]}],
-    rabbit_ct_helpers:run_setup_steps(
-      Config,
-      [fun(Conf) -> rabbit_ct_helpers:merge_app_env(Conf, Env) end] ++
-      rabbit_ct_broker_helpers:setup_steps() ++
-      rabbit_ct_client_helpers:setup_steps()).
+    Config = rabbit_ct_helpers:run_setup_steps(
+               Config1,
+               [fun(Conf) -> rabbit_ct_helpers:merge_app_env(Conf, Env) end] ++
+               rabbit_ct_broker_helpers:setup_steps() ++
+               rabbit_ct_client_helpers:setup_steps()),
+    ok = rabbit_ct_broker_helpers:enable_feature_flag(Config, mqtt_v5),
+    Config.
 
 end_per_group(G, Config)
   when G =:= v4;
diff --git a/deps/rabbitmq_mqtt/test/shared_SUITE.erl b/deps/rabbitmq_mqtt/test/shared_SUITE.erl
index 7feb768820..37bccf0bcb 100644
--- a/deps/rabbitmq_mqtt/test/shared_SUITE.erl
+++ b/deps/rabbitmq_mqtt/test/shared_SUITE.erl
@@ -120,6 +120,7 @@ cluster_size_1_tests() ->
      ,default_queue_type
      ,incoming_message_interceptors
      ,utf8
+     ,bind_exchange_to_exchange
     ].
 
 cluster_size_3_tests() ->
@@ -475,7 +476,8 @@ events(Config) ->
                                {pid, ConnectionPid}],
     assert_event_prop(ExpectedConnectionProps, E1),
 
-    {ok, _, _} = emqtt:subscribe(C, <<"TopicA">>, qos0),
+    Qos = 0,
+    {ok, _, _} = emqtt:subscribe(C, <<"TopicA">>, Qos),
 
     QueueNameBin = <<"mqtt-subscription-", ClientId/binary, "qos0">>,
     QueueName = {resource, <<"/">>, queue, QueueNameBin},
@@ -503,12 +505,16 @@ events(Config) ->
                        {arguments, []}],
                       E2),
     assert_event_type(binding_created, E3),
+    ExpectedBindingArgs = case ?config(mqtt_version, Config) of
+                              v5 -> [{mqtt_subscription_opts, Qos, false, false, 0, undefined}];
+                              _ -> []
+                          end,
     assert_event_prop([{source_name, <<"amq.topic">>},
                        {source_kind, exchange},
                        {destination_name, QueueNameBin},
                        {destination_kind, queue},
                        {routing_key, <<"TopicA">>},
-                       {arguments, []}],
+                       {arguments, ExpectedBindingArgs}],
                       E3),
 
     {ok, _, _} = emqtt:unsubscribe(C, <<"TopicA">>),
@@ -1357,6 +1363,7 @@ trace(Config) ->
     {ok, _, [0]} = emqtt:subscribe(Sub, Topic, qos0),
     {ok, _} = emqtt:publish(Pub, Topic, Payload, qos1),
     ok = expect_publishes(Sub, Topic, [Payload]),
+    timer:sleep(10),
 
     {#'basic.get_ok'{routing_key = <<"publish.amq.topic">>},
      #amqp_msg{props = #'P_basic'{headers = PublishHeaders},
@@ -1527,6 +1534,30 @@ utf8(Config) ->
     ok = expect_publishes(C, Topic, [<<"msg">>]),
     ok = emqtt:disconnect(C).
 
+bind_exchange_to_exchange(Config) ->
+    SourceX = <<"amq.topic">>,
+    DestinationX = <<"destination">>,
+    Q = <<"q">>,
+    Ch = rabbit_ct_client_helpers:open_channel(Config),
+    #'exchange.declare_ok'{} = amqp_channel:call(Ch, #'exchange.declare'{exchange = DestinationX,
+                                                                         durable = true,
+                                                                         auto_delete = true}),
+    #'queue.declare_ok'{} = amqp_channel:call(Ch, #'queue.declare'{queue = Q,
+                                                                   durable = true}),
+    #'queue.bind_ok'{} = amqp_channel:call(Ch, #'queue.bind'{exchange = DestinationX,
+                                                             queue = Q,
+                                                             routing_key = <<"a.b">>}),
+    #'exchange.bind_ok'{} = amqp_channel:call(Ch, #'exchange.bind'{destination = DestinationX,
+                                                                   source = SourceX,
+                                                                   routing_key = <<"*.b">>}),
+    C = connect(?FUNCTION_NAME, Config),
+    %% Message should be routed as follows: SourceX -> DestinationX -> Q
+    {ok, _} = emqtt:publish(C, <<"a/b">>, <<"msg">>, qos1),
+    eventually(?_assertMatch({#'basic.get_ok'{}, #amqp_msg{payload = <<"msg">>}},
+                             amqp_channel:call(Ch, #'basic.get'{queue = Q}))),
+    #'queue.delete_ok'{} = amqp_channel:call(Ch, #'queue.delete'{queue = Q}),
+    ok = emqtt:disconnect(C).
+
 %% -------------------------------------------------------------------
 %% Internal helpers
 %% -------------------------------------------------------------------
diff --git a/deps/rabbitmq_mqtt/test/util.erl b/deps/rabbitmq_mqtt/test/util.erl
index 2b4d27c085..206cf8813c 100644
--- a/deps/rabbitmq_mqtt/test/util.erl
+++ b/deps/rabbitmq_mqtt/test/util.erl
@@ -135,29 +135,15 @@ await_exit(Pid, Reason) ->
         20_000 -> ct:fail({missing_exit, Pid})
     end.
 
-%% In mixed version, we skip MQTT v5 tests if the old version does not support it.
 maybe_skip_v5(Config) ->
-    case {?config(mqtt_version, Config),
-          ?config(rmq_nodes_count, Config)} of
-        {v5, C} when C > 1 ->
-            ClientId = ?FUNCTION_NAME,
-            %% We should always be able to connect to the 1st node as it runs the latest version.
-            C1 = connect(ClientId, Config, 0, []),
-            ok = emqtt:disconnect(C1),
-            %% Check whether we can connect to the 2nd node.
-            {C2, Connect} = start_client(ClientId, Config, 1, [{connect_timeout, 3}]),
-            true = unlink(C2),
-            Skip = {skip, "The 2nd node in the cluster does not support MQTT 5.0"},
-            try Connect(C2) of
-                {ok, _} ->
-                    ok = emqtt:disconnect(C2),
+    case ?config(mqtt_version, Config) of
+        v5 ->
+            case rabbit_ct_broker_helpers:is_feature_flag_supported(Config, mqtt_v5) of
+                true ->
+                    ok = rabbit_ct_broker_helpers:enable_feature_flag(Config, mqtt_v5),
                     Config;
-                {error, _} ->
-                    %% MQTT
-                    Skip
-            catch exit:_ ->
-                      %% Web MQTT
-                      Skip
+                false ->
+                    {skip, "Feature flag mqtt_v5 is not supported by the entire cluster"}
             end;
         _ ->
             Config
@@ -193,9 +179,13 @@ start_client(ClientId, Config, Node, AdditionalOpts) ->
              [{ws_path, "/ws"}],
              fun emqtt:ws_connect/1}
     end,
+    ProtoVer = proplists:get_value(
+                 proto_ver,
+                 AdditionalOpts,
+                 rabbit_ct_helpers:get_config(Config, mqtt_version, v4)),
     Options = [{host, "localhost"},
                {port, Port},
-               {proto_ver, rabbit_ct_helpers:get_config(Config, mqtt_version, v4)},
+               {proto_ver, ProtoVer},
                {clientid, rabbit_data_coercion:to_binary(ClientId)}
               ] ++ WsOpts ++ AdditionalOpts,
     {ok, C} = emqtt:start_link(Options),
diff --git a/deps/rabbitmq_mqtt/test/v5_SUITE.erl b/deps/rabbitmq_mqtt/test/v5_SUITE.erl
index 5c57b4f900..71f68bd289 100644
--- a/deps/rabbitmq_mqtt/test/v5_SUITE.erl
+++ b/deps/rabbitmq_mqtt/test/v5_SUITE.erl
@@ -12,9 +12,10 @@
 
 -include_lib("common_test/include/ct.hrl").
 -include_lib("eunit/include/eunit.hrl").
+-include_lib("amqp_client/include/amqp_client.hrl").
 
 -import(util,
-        [
+        [all_connection_pids/1,
          start_client/4,
          connect/2, connect/3, connect/4,
          assert_message_expiry_interval/2,
@@ -32,6 +33,7 @@
 %% defined in MQTT v5 (not in v4 or v3)
 -define(RC_SUCCESS, 16#00).
 -define(RC_NO_SUBSCRIPTION_EXISTED, 16#11).
+-define(RC_UNSPECIFIED_ERROR, 16#80).
 
 all() ->
     [{group, mqtt},
@@ -40,12 +42,12 @@ all() ->
 groups() ->
     [
      {mqtt, [],
-      [{cluster_size_1, [shuffle], cluster_size_1_tests()}
-       % {cluster_size_3, [shuffle], cluster_size_3_tests()}
+      [{cluster_size_1, [shuffle], cluster_size_1_tests()},
+       {cluster_size_3, [shuffle], cluster_size_3_tests()}
       ]},
      {web_mqtt, [],
-      [{cluster_size_1, [shuffle], cluster_size_1_tests()}
-       % {cluster_size_3, [shuffle], cluster_size_3_tests()}
+      [{cluster_size_1, [shuffle], cluster_size_1_tests()},
+       {cluster_size_3, [shuffle], cluster_size_3_tests()}
       ]}
     ].
 
@@ -72,14 +74,34 @@ cluster_size_1_tests() ->
      client_receive_maximum_min,
      client_receive_maximum_large,
      unsubscribe_success,
-     unsubscribe_topic_not_found
+     unsubscribe_topic_not_found,
+     subscription_option_no_local,
+     subscription_option_no_local_wildcards,
+     subscription_option_retain_as_published,
+     subscription_option_retain_as_published_wildcards,
+     subscription_identifier,
+     subscription_identifier_amqp091,
+     subscription_identifier_at_most_once_dead_letter,
+     at_most_once_dead_letter_detect_cycle,
+     subscription_options_persisted,
+     subscription_options_modify,
+     subscription_options_modify_qos1,
+     subscription_options_modify_qos0,
+     session_upgrade_v3_v5_qos1,
+     session_upgrade_v3_v5_qos0,
+     compatibility_v3_v5,
+     session_upgrade_v3_v5_unsubscribe,
+     session_upgrade_v4_v5_no_queue_bind_permission,
+     amqp091_cc_header
     ].
 
-% cluster_size_3_tests() ->
-%     [].
+cluster_size_3_tests() ->
+    [session_migrate_v3_v5,
+     session_takeover_v3_v5
+    ].
 
 suite() ->
-    [{timetrap, {minutes, 1}}].
+    [{timetrap, {minutes, 10}}].
 
 %% -------------------------------------------------------------------
 %% Testsuite setup/teardown.
@@ -118,7 +140,13 @@ init_per_group(Group, Config0) ->
                Config2,
                rabbit_ct_broker_helpers:setup_steps() ++
                rabbit_ct_client_helpers:setup_steps()),
-    util:maybe_skip_v5(Config).
+    case Group of
+        cluster_size_1 ->
+            ok = rabbit_ct_broker_helpers:enable_feature_flag(Config, mqtt_v5);
+        cluster_size_3 ->
+            ok
+    end,
+    Config.
 
 end_per_group(G, Config)
   when G =:= cluster_size_1;
@@ -138,9 +166,13 @@ init_per_testcase(T, Config)
     {ok, Default} = rpc(Config, application, get_env, [?APP, Par]),
     ok = rpc(Config, application, set_env, [?APP, Par, infinity]),
     Config1 = rabbit_ct_helpers:set_config(Config, {Par, Default}),
-    rabbit_ct_helpers:testcase_started(Config1, T);
+    init_per_testcase0(T, Config1);
+init_per_testcase(T, Config) ->
+    init_per_testcase0(T, Config).
 
-init_per_testcase(Testcase, Config) ->
+init_per_testcase0(Testcase, Config) ->
+    Nodes = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+    [ok = rabbit_ct_broker_helpers:enable_plugin(Config, N, rabbitmq_web_mqtt) || N <- Nodes],
     rabbit_ct_helpers:testcase_started(Config, Testcase).
 
 end_per_testcase(T, Config)
@@ -179,6 +211,7 @@ client_set_max_packet_size_connack(Config) ->
     {C, Connect} = start_client(?FUNCTION_NAME, Config, 0,
                                 [{properties, #{'Maximum-Packet-Size' => 2}},
                                  {connect_timeout, 1}]),
+    unlink(C),
     %% We expect the server to drop the CONNACK packet because it's larger than 2 bytes.
     ?assertEqual({error, connack_timeout}, Connect(C)).
 
@@ -457,7 +490,7 @@ client_rejects_publish(Config) ->
     receive {publish, #{payload := Payload,
                         packet_id := PacketId}} ->
                 %% Negatively ack the PUBLISH.
-                emqtt:puback(C, PacketId, _UnspecifiedError = 16#80)
+                emqtt:puback(C, PacketId, ?RC_UNSPECIFIED_ERROR)
     after 1000 ->
               ct:fail("did not receive PUBLISH")
     end,
@@ -519,22 +552,651 @@ client_receive_maximum_large(Config) ->
     ok = emqtt:disconnect(C).
 
 unsubscribe_success(Config) ->
-    ClientId = atom_to_binary(?FUNCTION_NAME),
-    C = connect(ClientId, Config),
+    C = connect(?FUNCTION_NAME, Config),
     {ok, _, [1]} = emqtt:subscribe(C, <<"topic/1">>, qos1),
     {ok, _, [0]} = emqtt:subscribe(C, <<"topic/0">>, qos0),
-    {ok, _, ReasonCodes} = emqtt:unsubscribe(C, [<<"topic/1">>, <<"topic/0">>]),
-    ?assertEqual([?RC_SUCCESS, ?RC_SUCCESS], ReasonCodes),
+    ?assertMatch({ok, _, [?RC_SUCCESS, ?RC_SUCCESS]},
+                 emqtt:unsubscribe(C, [<<"topic/1">>, <<"topic/0">>])),
     ok = emqtt:disconnect(C).
 
 unsubscribe_topic_not_found(Config) ->
-    ClientId = atom_to_binary(?FUNCTION_NAME),
-    C = connect(ClientId, Config),
+    C = connect(?FUNCTION_NAME, Config),
     {ok, _, [1]} = emqtt:subscribe(C, <<"topic/1">>, qos1),
-    {ok, _, ReasonCodes} = emqtt:unsubscribe(C, [<<"topic/1">>, <<"topic/0">>]),
-    ?assertEqual([?RC_SUCCESS, ?RC_NO_SUBSCRIPTION_EXISTED], ReasonCodes),
+    ?assertMatch({ok, _, [?RC_SUCCESS, ?RC_NO_SUBSCRIPTION_EXISTED]},
+                 emqtt:unsubscribe(C, [<<"topic/1">>, <<"topic/0">>])),
+    ok = emqtt:disconnect(C).
+
+subscription_option_no_local(Config) ->
+    C = connect(?FUNCTION_NAME, Config),
+    Other = connect(<<"other">>, Config),
+    {ok, _, [0, 0]} = emqtt:subscribe(C, [{<<"t/1">>, [{nl, true}]},
+                                          {<<"t/2">>, [{nl, false}]}]),
+    {ok, _, [0]} = emqtt:subscribe(Other, [{<<"t/1">>, [{nl, true}]}]),
+    ok = emqtt:publish(C, <<"t/1">>, <<"m1">>),
+    ok = emqtt:publish(C, <<"t/2">>, <<"m2">>),
+    ok = expect_publishes(Other, <<"t/1">>, [<<"m1">>]),
+    ok = expect_publishes(C, <<"t/2">>, [<<"m2">>]),
+    %% We expect C to not receive m1.
+    assert_nothing_received(),
+    ok = emqtt:disconnect(C),
+    ok = emqtt:disconnect(Other).
+
+subscription_option_no_local_wildcards(Config) ->
+    C = connect(?FUNCTION_NAME, Config),
+    {ok, _, [0, 0, 0, 0]} =
+    emqtt:subscribe(C, [{<<"+/1">>, [{nl, true}]},
+                        {<<"t/1/#">>, [{nl, true}]},
+                        {<<"+/2">>, [{nl, true}]},
+                        {<<"t/2/#">>, [{nl, false}]}]),
+    %% Matches the first two subscriptions.
+    %% All matching subscriptions have the No Local option set.
+    %% Therefore, we should not receive m1.
+    ok = emqtt:publish(C, <<"t/1">>, <<"m1">>),
+    %% Matches the last two subscriptions.
+    %% Not all matching subscriptions have the No Local option set.
+    %% Therefore, we should receive m2.
+    ok = emqtt:publish(C, <<"t/2">>, <<"m2">>),
+    ok = expect_publishes(C, <<"t/2">>, [<<"m2">>]),
+    assert_nothing_received(),
+    ok = emqtt:disconnect(C).
+
+subscription_option_retain_as_published(Config) ->
+    C1 = connect(<<"c1">>, Config),
+    C2 = connect(<<"c2">>, Config),
+    {ok, _, [0, 0]} = emqtt:subscribe(C1, [{<<"t/1">>, [{rap, true}]},
+                                           {<<"t/2">>, [{rap, false}]}]),
+    {ok, _, [0]} = emqtt:subscribe(C2, [{<<"t/1">>, [{rap, true}]}]),
+    ok = emqtt:publish(C1, <<"t/1">>, <<"m1">>, [{retain, true}]),
+    ok = emqtt:publish(C1, <<"t/2">>, <<"m2">>, [{retain, true}]),
+    receive {publish, #{client_pid := C1,
+                        topic := <<"t/1">>,
+                        payload := <<"m1">>,
+                        retain := true}} -> ok
+    after 1000 -> ct:fail("did not receive m1")
+    end,
+    receive {publish, #{client_pid := C1,
+                        topic := <<"t/2">>,
+                        payload := <<"m2">>,
+                        retain := false}} -> ok
+    after 1000 -> ct:fail("did not receive m2")
+    end,
+    receive {publish, #{client_pid := C2,
+                        topic := <<"t/1">>,
+                        payload := <<"m1">>,
+                        retain := true}} -> ok
+    after 1000 -> ct:fail("did not receive m1")
+    end,
+    ok = emqtt:publish(C1, <<"t/1">>, <<"">>, [{retain, true}]),
+    ok = emqtt:publish(C1, <<"t/2">>, <<"">>, [{retain, true}]),
+    ok = emqtt:disconnect(C1),
+    ok = emqtt:disconnect(C2).
+
+subscription_option_retain_as_published_wildcards(Config) ->
+    C = connect(?FUNCTION_NAME, Config),
+    {ok, _, [0, 0, 0, 0]} = emqtt:subscribe(C, [{<<"+/1">>, [{rap, false}]},
+                                                {<<"t/1/#">>, [{rap, false}]},
+                                                {<<"+/2">>, [{rap, false}]},
+                                                {<<"t/2/#">>, [{rap, true}]}]),
+    %% Matches the first two subscriptions.
+    ok = emqtt:publish(C, <<"t/1">>, <<"m1">>, [{retain, true}]),
+    %% Matches the last two subscriptions.
+    ok = emqtt:publish(C, <<"t/2">>, <<"m2">>, [{retain, true}]),
+    receive {publish, #{topic := <<"t/1">>,
+                        payload := <<"m1">>,
+                        %% No matching subscription has the
+                        %% Retain As Published option set.
+                        retain := false}} -> ok
+    after 1000 -> ct:fail("did not receive m1")
+    end,
+    receive {publish, #{topic := <<"t/2">>,
+                        payload := <<"m2">>,
+                        %% (At least) one matching subscription has the
+                        %% Retain As Published option set.
+                        retain := true}} -> ok
+    after 1000 -> ct:fail("did not receive m2")
+    end,
+    ok = emqtt:publish(C, <<"t/1">>, <<"">>, [{retain, true}]),
+    ok = emqtt:publish(C, <<"t/2">>, <<"">>, [{retain, true}]),
     ok = emqtt:disconnect(C).
 
+subscription_identifier(Config) ->
+    C1 = connect(<<"c1">>, Config),
+    C2 = connect(<<"c2">>, Config),
+    {ok, _, [0, 0]} = emqtt:subscribe(C1, #{'Subscription-Identifier' => 1}, [{<<"t/1">>, []},
+                                                                              {<<"+/1">>, []}]),
+    {ok, _, [0]} = emqtt:subscribe(C2, #{'Subscription-Identifier' => 1}, [{<<"t/2">>, []}]),
+    {ok, _, [0, 0]} = emqtt:subscribe(C2, #{'Subscription-Identifier' => 16#fffffff}, [{<<"+/2">>, []},
+                                                                                       {<<"t/3">>, []}]),
+    {ok, _, [0]} = emqtt:subscribe(C2, #{}, [{<<"t/2/#">>, []}]),
+    ok = emqtt:publish(C1, <<"t/1">>, <<"m1">>),
+    ok = emqtt:publish(C1, <<"t/2">>, <<"m2">>),
+    ok = emqtt:publish(C1, <<"t/3">>, <<"m3">>),
+    ok = emqtt:publish(C1, <<"t/2/xyz">>, <<"m4">>),
+    receive {publish,
+             #{client_pid := C1,
+               topic := <<"t/1">>,
+               payload := <<"m1">>,
+               %% "It is possible that the Client made several subscriptions which match a publication
+               %% and that it used the same identifier for more than one of them. In this case the
+               %% PUBLISH packet will carry multiple identical Subscription Identifiers." [v5 3.3.4]
+               properties := #{'Subscription-Identifier' := [1, 1]}}} -> ok
+    after 1000 -> ct:fail("did not receive m1")
+    end,
+    receive {publish,
+             #{client_pid := C2,
+               topic := <<"t/2">>,
+               payload := <<"m2">>,
+               properties := #{'Subscription-Identifier' := Ids}}} ->
+                %% "If the Server sends a single copy of the message it MUST include in the PUBLISH
+                %% packet the Subscription Identifiers for all matching subscriptions which have a
+                %% Subscription Identifiers, their order is not significant [MQTT-3.3.4-4]." [v5 3.3.4]
+                ?assertEqual([1, 16#fffffff], lists:sort(Ids))
+    after 1000 -> ct:fail("did not receive m2")
+    end,
+    receive {publish,
+             #{client_pid := C2,
+               topic := <<"t/3">>,
+               payload := <<"m3">>,
+               properties := #{'Subscription-Identifier' := 16#fffffff}}} -> ok
+    after 1000 -> ct:fail("did not receive m3")
+    end,
+    receive {publish,
+             #{client_pid := C2,
+               topic := <<"t/2/xyz">>,
+               payload := <<"m4">>,
+               properties := Props}} ->
+                ?assertNot(maps:is_key('Subscription-Identifier', Props))
+    after 1000 -> ct:fail("did not receive m4")
+    end,
+    assert_nothing_received(),
+    ok = emqtt:disconnect(C1),
+    ok = emqtt:disconnect(C2).
+
+subscription_identifier_amqp091(Config) ->
+    C1 = connect(<<"sub 1">>, Config),
+    C2 = connect(<<"sub 2">>, Config),
+    {ok, _, [1]} = emqtt:subscribe(C1, #{'Subscription-Identifier' => 1}, [{<<"a/+">>, [{qos, 1}]}]),
+    {ok, _, [1]} = emqtt:subscribe(C2, #{'Subscription-Identifier' => 16#fffffff}, [{<<"a/b">>, [{qos, 1}]}]),
+    Ch = rabbit_ct_client_helpers:open_channel(Config),
+
+    %% Test routing to a single queue.
+    amqp_channel:call(Ch, #'basic.publish'{exchange = <<"amq.topic">>,
+                                           routing_key = <<"a.a">>},
+                      #amqp_msg{payload = <<"m1">>}),
+    receive {publish,
+             #{client_pid := C1,
+               topic := <<"a/a">>,
+               payload := <<"m1">>,
+               properties := #{'Subscription-Identifier' := 1}}} -> ok
+    after 1000 -> ct:fail("did not receive message m1")
+    end,
+
+    %% Test routing to multiple queues.
+    amqp_channel:call(Ch, #'basic.publish'{exchange = <<"amq.topic">>,
+                                           routing_key = <<"a.b">>},
+                      #amqp_msg{payload = <<"m2">>}),
+    receive {publish,
+             #{client_pid := C1,
+               topic := <<"a/b">>,
+               payload := <<"m2">>,
+               properties := #{'Subscription-Identifier' := 1}}} -> ok
+    after 1000 -> ct:fail("did not receive message m2")
+    end,
+    receive {publish,
+             #{client_pid := C2,
+               topic := <<"a/b">>,
+               payload := <<"m2">>,
+               properties := #{'Subscription-Identifier' := 16#fffffff}}} -> ok
+    after 1000 -> ct:fail("did not receive message m2")
+    end,
+
+    ok = emqtt:disconnect(C1),
+    ok = emqtt:disconnect(C2),
+    ok = rabbit_ct_client_helpers:close_channels_and_connection(Config, 0).
+
+subscription_identifier_at_most_once_dead_letter(Config) ->
+    C = connect(?FUNCTION_NAME, Config),
+    {ok, _, [1]} = emqtt:subscribe(C, #{'Subscription-Identifier' => 1}, [{<<"dead letter/#">>, [{qos, 1}]}]),
+
+    Ch = rabbit_ct_client_helpers:open_channel(Config),
+    QArgs = [{<<"x-dead-letter-exchange">>, longstr, <<"amq.topic">>},
+             {<<"x-dead-letter-routing-key">>, longstr, <<"dead letter.a">>}],
+    #'queue.declare_ok'{} = amqp_channel:call(Ch, #'queue.declare'{queue = <<"source queue">>,
+                                                                   durable = true,
+                                                                   exclusive = true,
+                                                                   arguments = QArgs}),
+    amqp_channel:call(Ch, #'basic.publish'{routing_key = <<"source queue">>},
+                      #amqp_msg{payload = <<"msg">>,
+                                props = #'P_basic'{expiration = <<"0">>}}),
+    receive {publish,
+             #{client_pid := C,
+               topic := <<"dead letter/a">>,
+               payload := <<"msg">>,
+               properties := #{'Subscription-Identifier' := 1}}} -> ok
+    after 1000 -> ct:fail("did not receive msg")
+    end,
+    ok = emqtt:disconnect(C),
+    ok = rabbit_ct_client_helpers:close_channels_and_connection(Config, 0).
+
+at_most_once_dead_letter_detect_cycle(Config) ->
+    NumExpiredBefore = dead_letter_metric(messages_dead_lettered_expired_total, Config, at_most_once),
+    SubClientId = Payload = PolicyName = atom_to_binary(?FUNCTION_NAME),
+    ok = rabbit_ct_broker_helpers:set_policy(
+           Config, 0, PolicyName, <<"mqtt-subscription-", SubClientId/binary, "qos1">>, <<"queues">>,
+           %% Create dead letter cycle: qos1 queue -> topic exchange -> qos1 queue
+           [{<<"dead-letter-exchange">>, <<"amq.topic">>},
+            {<<"message-ttl">>, 1}]),
+    Sub1 = connect(SubClientId, Config, non_clean_sess_opts()),
+    {ok, _, [1]} = emqtt:subscribe(Sub1, #{'Subscription-Identifier' => 10}, [{<<"+/b">>, [{qos, 1}]}]),
+    ok = emqtt:disconnect(Sub1),
+
+    Pub = connect(<<"publisher">>, Config),
+    {ok, _} = emqtt:publish(Pub, <<"a/b">>, Payload, qos1),
+    ok = emqtt:disconnect(Pub),
+    %% Given our subscribing client is disconnected, the message should be dead lettered after 1 ms.
+    %% However, due to the dead letter cycle, we expect the message to be dropped.
+    timer:sleep(5),
+    Sub2 = connect(SubClientId, Config, [{clean_start, false}]),
+    assert_nothing_received(),
+    %% Double check that the message was indeed (exactly once) dead lettered.
+    NumExpired = dead_letter_metric(messages_dead_lettered_expired_total,
+                                    Config, at_most_once) - NumExpiredBefore,
+    ?assertEqual(1, NumExpired),
+    ok = emqtt:disconnect(Sub2),
+    ok = rabbit_ct_broker_helpers:clear_policy(Config, 0, PolicyName).
+
+%% Tests that the session state in the server includes subscription options
+%% and subscription identifiers and that this session state is persisted.
+subscription_options_persisted(Config) ->
+    ClientId = ?FUNCTION_NAME,
+    C1 = connect(ClientId, Config, non_clean_sess_opts()),
+    {ok, _, [0, 1]} = emqtt:subscribe(C1, #{'Subscription-Identifier' => 99},
+                                      [{<<"t1">>, [{nl, true}, {rap, false}, {qos, 0}]},
+                                       {<<"t2">>, [{nl, false}, {rap, true}, {qos, 1}]}]),
+    unlink(C1),
+    ok = rabbit_ct_broker_helpers:restart_node(Config, 0),
+    C2 = connect(ClientId, Config, [{clean_start, false}]),
+    ok = emqtt:publish(C2, <<"t1">>, <<"m1">>),
+    ok = emqtt:publish(C2, <<"t2">>, <<"m2">>, [{retain, true}]),
+    receive {publish,
+             #{client_pid := C2,
+               payload := <<"m2">>,
+               retain := true,
+               qos := 0,
+               properties := #{'Subscription-Identifier' := 99}}} -> ok
+    after 1000 -> ct:fail("did not receive m2")
+    end,
+    assert_nothing_received(),
+    ok = emqtt:publish(C2, <<"t2">>, <<"">>, [{retain, true}]),
+    ok = emqtt:disconnect(C2).
+
+%% "If a Server receives a SUBSCRIBE packet containing a Topic Filter that is identical to a Non‑shared
+%% Subscription’s Topic Filter for the current Session, then it MUST replace that existing Subscription
+%% with a new Subscription [MQTT-3.8.4-3]. The Topic Filter in the new Subscription will be identical
+%% to that in the previous Subscription, although its Subscription Options could be different."
+%%
+%% "The Subscription Identifiers are part of the Session State in the Server and are returned to the
+%% Client receiving a matching PUBLISH packet. They are removed from the Server’s Session State when the
+%% Server receives an UNSUBSCRIBE packet, when the Server receives a SUBSCRIBE packet from the Client for
+%% the same Topic Filter but with a different Subscription Identifier or with no Subscription Identifier,
+%% or when the Server sends Session Present 0 in a CONNACK packet" [v5 3.8.4]
+subscription_options_modify(Config) ->
+    Topic = ClientId = atom_to_binary(?FUNCTION_NAME),
+    C = connect(ClientId, Config),
+
+    {ok, _, [0]} = emqtt:subscribe(C, #{'Subscription-Identifier' => 1}, Topic, [{nl, true}]),
+    {ok, _} = emqtt:publish(C, Topic, <<"m1">>, qos1),
+    assert_nothing_received(),
+
+    %% modify No Local
+    {ok, _, [0]} = emqtt:subscribe(C, #{'Subscription-Identifier' => 1}, Topic, [{nl, false}]),
+    {ok, _} = emqtt:publish(C, Topic, <<"m2">>, qos1),
+    receive {publish, #{payload := <<"m2">>,
+                        qos := 0 }} -> ok
+    after 1000 -> ct:fail("did not receive m2")
+    end,
+
+    %% modify QoS
+    {ok, _, [1]} = emqtt:subscribe(C, #{'Subscription-Identifier' => 1}, Topic, qos1),
+    {ok, _} = emqtt:publish(C, Topic, <<"m3">>, qos1),
+    receive {publish, #{payload := <<"m3">>,
+                        qos := 1,
+                        properties := #{'Subscription-Identifier' := 1}}} -> ok
+    after 1000 -> ct:fail("did not receive m3")
+    end,
+
+    %% modify Subscription Identifier
+    {ok, _, [1]} = emqtt:subscribe(C, #{'Subscription-Identifier' => 2}, Topic, qos1),
+    {ok, _} = emqtt:publish(C, Topic, <<"m4">>, qos1),
+    receive {publish, #{payload := <<"m4">>,
+                        properties := #{'Subscription-Identifier' := 2}}} -> ok
+    after 1000 -> ct:fail("did not receive m4")
+    end,
+
+    %% remove Subscription Identifier
+    {ok, _, [1]} = emqtt:subscribe(C, Topic, qos1),
+    {ok, _} = emqtt:publish(C, Topic, <<"m5">>, [{retain, true}, {qos, 1}]),
+    receive {publish, #{payload := <<"m5">>,
+                        retain := false,
+                        properties := Props}} when map_size(Props) =:= 0 -> ok
+    after 1000 -> ct:fail("did not receive m5")
+    end,
+
+    %% modify Retain As Published
+    {ok, _, [1]} = emqtt:subscribe(C, Topic, [{rap, true}, {qos, 1}]),
+    receive {publish, #{payload := <<"m5">>,
+                        retain := true}} -> ok
+    after 1000 -> ct:fail("did not receive retained m5")
+    end,
+    {ok, _} = emqtt:publish(C, Topic, <<"m6">>, [{retain, true}, {qos, 1}]),
+    receive {publish, #{payload := <<"m6">>,
+                        retain := true}} -> ok
+    after 1000 -> ct:fail("did not receive m6")
+    end,
+
+    assert_nothing_received(),
+    ok = emqtt:publish(C, Topic, <<"">>, [{retain, true}]),
+    ok = emqtt:disconnect(C).
+
+%% "If a Server receives a SUBSCRIBE packet containing a Topic Filter that is identical to a
+%% Non‑shared Subscription’s Topic Filter for the current Session, then it MUST replace that
+%% existing Subscription with a new Subscription [MQTT-3.8.4-3]. The Topic Filter in the new
+%% Subscription will be identical to that in the previous Subscription, although its
+%% Subscription Options could be different. [...] Applicaton Messages MUST NOT be lost due
+%% to replacing the Subscription [MQTT-3.8.4-4]." [v5 3.8.4]
+%%
+%% This test ensures that messages are not lost when replacing a QoS 1 subscription.
+subscription_options_modify_qos1(Config) ->
+    subscription_options_modify_qos(1, Config).
+
+%% This test ensures that messages are received at most once
+%% when replacing a QoS 0 subscription.
+subscription_options_modify_qos0(Config) ->
+    subscription_options_modify_qos(0, Config).
+
+subscription_options_modify_qos(Qos, Config) ->
+    Topic = atom_to_binary(?FUNCTION_NAME),
+    Pub = connect(<<"publisher">>, Config),
+    Sub = connect(<<"subscriber">>, Config),
+    {ok, _, [Qos]} = emqtt:subscribe(Sub, Topic, Qos),
+    Sender = spawn_link(?MODULE, send, [self(), Pub, Topic, 0]),
+    receive {publish, #{payload := <<"1">>,
+                        properties := Props}} ->
+                ?assertEqual(0, maps:size(Props))
+    after 1000 -> ct:fail("did not receive 1")
+    end,
+    %% Replace subscription while another client is sending messages.
+    {ok, _, [Qos]} = emqtt:subscribe(Sub, #{'Subscription-Identifier' => 1}, Topic, Qos),
+    Sender ! stop,
+    NumSent = receive {N, Sender} -> N
+              after 1000 -> ct:fail("could not stop publisher")
+              end,
+    ct:pal("Publisher sent ~b messages", [NumSent]),
+    LastExpectedPayload = integer_to_binary(NumSent),
+    receive {publish, #{payload := LastExpectedPayload,
+                        qos := Qos,
+                        client_pid := Sub,
+                        properties := #{'Subscription-Identifier' := 1}}} -> ok
+    after 1000 -> ct:fail("did not receive ~s", [LastExpectedPayload])
+    end,
+    case Qos of
+        0 ->
+            assert_received_no_duplicates();
+        1 ->
+            ExpectedPayloads = [integer_to_binary(I) || I <- lists:seq(2, NumSent - 1)],
+            ok = util:expect_publishes(Sub, Topic, ExpectedPayloads)
+    end,
+    ok = emqtt:disconnect(Pub),
+    ok = emqtt:disconnect(Sub).
+
+%% Tests that no message is lost when upgrading a session
+%% with QoS 1 subscription from v3 to v5.
+session_upgrade_v3_v5_qos1(Config) ->
+    session_upgrade_v3_v5_qos(1, Config).
+
+%% Tests that each message is received at most once
+%% when upgrading a session with QoS 0 subscription from v3 to v5.
+session_upgrade_v3_v5_qos0(Config) ->
+    session_upgrade_v3_v5_qos(0, Config).
+
+session_upgrade_v3_v5_qos(Qos, Config) ->
+    ClientId = Topic = atom_to_binary(?FUNCTION_NAME),
+    Pub = connect(<<"publisher">>, Config),
+    Subv3 = connect(ClientId, Config, [{proto_ver, v3} | non_clean_sess_opts()]),
+    ?assertEqual(3, proplists:get_value(proto_ver, emqtt:info(Subv3))),
+    {ok, _, [Qos]} = emqtt:subscribe(Subv3, Topic, Qos),
+    Sender = spawn_link(?MODULE, send, [self(), Pub, Topic, 0]),
+    receive {publish, #{payload := <<"1">>,
+                        client_pid := Subv3}} -> ok
+    after 1000 -> ct:fail("did not receive 1")
+    end,
+    %% Upgrade session from v3 to v5 while another client is sending messages.
+    ok = emqtt:disconnect(Subv3),
+    Subv5 = connect(ClientId, Config, [{proto_ver, v5}, {clean_start, false}]),
+    ?assertEqual(5, proplists:get_value(proto_ver, emqtt:info(Subv5))),
+    Sender ! stop,
+    NumSent = receive {N, Sender} -> N
+              after 1000 -> ct:fail("could not stop publisher")
+              end,
+    ct:pal("Publisher sent ~b messages", [NumSent]),
+    LastExpectedPayload = integer_to_binary(NumSent),
+    receive {publish, #{payload := LastExpectedPayload,
+                        qos := Qos,
+                        client_pid := Subv5}} -> ok
+    after 1000 -> ct:fail("did not receive ~s", [LastExpectedPayload])
+    end,
+    case Qos of
+        0 ->
+            assert_received_no_duplicates();
+        1 ->
+            ExpectedPayloads = [integer_to_binary(I) || I <- lists:seq(2, NumSent - 1)],
+            ok = util:expect_publishes(Subv5, Topic, ExpectedPayloads)
+    end,
+    ok = emqtt:disconnect(Pub),
+    ok = emqtt:disconnect(Subv5).
+
+send(Parent, Client, Topic, NumSent) ->
+    receive stop ->
+                Parent ! {NumSent, self()}
+    after 0 ->
+              N = NumSent + 1,
+              {ok, _} = emqtt:publish(Client, Topic, integer_to_binary(N), qos1),
+              send(Parent, Client, Topic, N)
+    end.
+
+assert_received_no_duplicates() ->
+    assert_received_no_duplicates0(#{}).
+
+assert_received_no_duplicates0(Received) ->
+    receive {publish, #{payload := P}} ->
+                case maps:is_key(P, Received) of
+                    true -> ct:fail("Received ~p twice", [P]);
+                    false -> assert_received_no_duplicates0(maps:put(P, ok, Received))
+                end
+    after 500 ->
+              %% Check that we received at least one message.
+              ?assertNotEqual(0, maps:size(Received))
+    end.
+
+compatibility_v3_v5(Config) ->
+    Cv3 = connect(<<"client v3">>, Config, [{proto_ver, v3}]),
+    Cv5 = connect(<<"client v5">>, Config, [{proto_ver, v5}]),
+    %% Sanity check that versions were set correctly.
+    ?assertEqual(3, proplists:get_value(proto_ver, emqtt:info(Cv3))),
+    ?assertEqual(5, proplists:get_value(proto_ver, emqtt:info(Cv5))),
+    {ok, _, [1]} = emqtt:subscribe(Cv3, <<"v3/#">>, qos1),
+    {ok, _, [1]} = emqtt:subscribe(Cv5, #{'Subscription-Identifier' => 99},
+                                   [{<<"v5/#">>, [{rap, true}, {qos, 1}]}]),
+    %% Send message in either direction.
+    {ok, _} = emqtt:publish(Cv5, <<"v3">>, <<"from v5">>, qos1),
+    {ok, _} = emqtt:publish(Cv3, <<"v5">>, <<"from v3">>, [{retain, true}, {qos, 1}]),
+    ok = expect_publishes(Cv3, <<"v3">>, [<<"from v5">>]),
+    receive {publish,
+             #{client_pid := Cv5,
+               topic := <<"v5">>,
+               payload := <<"from v3">>,
+               %% v5 features should work even when message comes from a v3 client.
+               retain := true,
+               properties := #{'Subscription-Identifier' := 99}}} -> ok
+    after 1000 -> ct:fail("did not receive from v3")
+    end,
+    ok = emqtt:publish(Cv3, <<"v5">>, <<"">>, [{retain, true}]),
+    ok = emqtt:disconnect(Cv3),
+    ok = emqtt:disconnect(Cv5).
+
+session_upgrade_v3_v5_unsubscribe(Config) ->
+    Topic = ClientId = atom_to_binary(?FUNCTION_NAME),
+    C1 = connect(ClientId, Config, [{proto_ver, v3} | non_clean_sess_opts()]),
+    ?assertEqual(3, proplists:get_value(proto_ver, emqtt:info(C1))),
+    {ok, _, [0]} = emqtt:subscribe(C1, Topic),
+    ok = emqtt:disconnect(C1),
+    %% Upgrade the session from v3 to v5.
+    C2 = connect(ClientId, Config, [{proto_ver, v5}, {clean_start, false}]),
+    ?assertEqual(5, proplists:get_value(proto_ver, emqtt:info(C2))),
+    ok = emqtt:publish(C2, Topic, <<"m1">>),
+    ok = expect_publishes(C2, Topic, [<<"m1">>]),
+    %% Unsubscribing in v5 should work.
+    ?assertMatch({ok, _, [?RC_SUCCESS]}, emqtt:unsubscribe(C2, Topic)),
+    ok = emqtt:publish(C2, Topic, <<"m2">>),
+    assert_nothing_received(),
+    ok = emqtt:disconnect(C2).
+
+session_upgrade_v4_v5_no_queue_bind_permission(Config) ->
+    Topic = ClientId = atom_to_binary(?FUNCTION_NAME),
+    C1 = connect(ClientId, Config, [{proto_ver, v4} | non_clean_sess_opts()]),
+    ?assertEqual(4, proplists:get_value(proto_ver, emqtt:info(C1))),
+    {ok, _, [0]} = emqtt:subscribe(C1, Topic),
+    ok = emqtt:disconnect(C1),
+
+    %% Revoking write access to queue will cause queue.bind to fail.
+    rabbit_ct_broker_helpers:set_permissions(Config, <<"guest">>, <<"/">>, <<".*">>, <<"">>, <<".*">>),
+    %% Upgrading the session from v4 to v5 should fail because it causes
+    %% queue.bind and queue.unbind (to change the binding arguments).
+    {C2, Connect} = start_client(ClientId, Config, 0, [{proto_ver, v5}, {clean_start, false}]),
+    unlink(C2),
+    ?assertEqual({error, {not_authorized, #{}}}, Connect(C2)),
+
+    %% Cleanup
+    rabbit_ct_broker_helpers:set_full_permissions(Config, <<"guest">>, <<"/">>),
+    C3 = connect(ClientId, Config),
+    ok = emqtt:disconnect(C3).
+
+amqp091_cc_header(Config) ->
+    C = connect(?FUNCTION_NAME, Config),
+    {ok, _, [1]} = emqtt:subscribe(C, #{'Subscription-Identifier' => 1}, [{<<"#">>, [{qos, 1}]}]),
+
+    Ch = rabbit_ct_client_helpers:open_channel(Config),
+    amqp_channel:call(
+      Ch, #'basic.publish'{exchange = <<"amq.topic">>,
+                           routing_key = <<"first.key">>},
+      #amqp_msg{payload = <<"msg">>,
+                props = #'P_basic'{
+                           headers = [{<<"CC">>, array,
+                                       [{longstr, <<"second.key">>}]}]}}),
+    %% Even though both routing key and CC header match the topic filter,
+    %% we expect to receive a single message (because only one message is sent)
+    %% and a single subscription identifier (because we created only one subscription).
+    receive {publish,
+             #{topic := <<"first/key">>,
+               payload := <<"msg">>,
+               properties := #{'Subscription-Identifier' := 1}}} -> ok
+    after 1000 -> ct:fail("did not receive msg")
+    end,
+    assert_nothing_received(),
+    ok = emqtt:disconnect(C).
+
+session_migrate_v3_v5(Config) ->
+    session_switch_v3_v5(Config, true).
+
+session_takeover_v3_v5(Config) ->
+    session_switch_v3_v5(Config, false).
+
+session_switch_v3_v5(Config, Disconnect) ->
+    V5Enabled = rabbit_ct_broker_helpers:is_feature_flag_enabled(Config, mqtt_v5),
+    [Server1, Server2, _] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+    Topic = ClientId = atom_to_binary(?FUNCTION_NAME),
+    %% Connect to old node in mixed version cluster.
+    C1 = connect(ClientId, Config, Server2, [{proto_ver, v3} | non_clean_sess_opts()]),
+    ?assertEqual(3, proplists:get_value(proto_ver, emqtt:info(C1))),
+    {ok, _, [1]} = emqtt:subscribe(C1, Topic, qos1),
+    case Disconnect of
+        true -> ok = emqtt:disconnect(C1);
+        false -> unlink(C1)
+    end,
+    monitor(process, C1),
+
+    case V5Enabled of
+        true ->
+            %% Upgrade session from v3 to v5.
+            C2 = connect(ClientId, Config, Server1, [{proto_ver, v5} | non_clean_sess_opts()]),
+            ?assertEqual(5, proplists:get_value(proto_ver, emqtt:info(C2))),
+            %% Subscription created with old v3 client should work.
+            {ok, _} = emqtt:publish(C2, Topic, <<"m1">>, qos1),
+            receive {publish,
+                     #{client_pid := C2,
+                       payload := <<"m1">>,
+                       qos := 1}} -> ok
+            after 1000 -> ct:fail("did not receive from m1")
+            end,
+            %% Modifying subscription with v5 specific feature should work.
+            {ok, _, [1]} = emqtt:subscribe(C2, Topic, [{nl, true}, {qos, 1}]),
+            {ok, _} = emqtt:publish(C2, Topic, <<"m2">>, qos1),
+            receive {publish, P} -> ct:fail("Unexpected local PUBLISH: ~p", [P])
+            after 500 -> ok
+            end,
+            case Disconnect of
+                true -> ok = emqtt:disconnect(C2);
+                false -> unlink(C2)
+            end;
+        false ->
+            ok
+    end,
+
+    %% If feature flag mqtt_v5 is
+    %% * enabled: downgrade session from v5 to v3
+    %% * disabled: in mixed version cluster connect to new node using v3
+    C3 = connect(ClientId, Config, Server1, [{proto_ver, v3} | non_clean_sess_opts()]),
+    ?assertEqual(3, proplists:get_value(proto_ver, emqtt:info(C3))),
+    receive {'DOWN', _, _, C1, _} -> ok
+    after 1000 -> ct:fail("C1 did not exit")
+    end,
+    case {Disconnect, V5Enabled} of
+        {false, true} ->
+            receive {disconnected, _RcSessionTakenOver = 142, #{}} -> ok
+            after 1000 -> ct:fail("missing DISCONNECT packet for C2")
+            end;
+        _ ->
+            ok
+    end,
+    %% We expect that v5 specific subscription feature does not apply
+    %% anymore when downgrading the session.
+    {ok, _} = emqtt:publish(C3, Topic, <<"m3">>, qos1),
+    receive {publish,
+             #{client_pid := C3,
+               payload := <<"m3">>,
+               qos := 1}} -> ok
+    after 1000 -> ct:fail("did not receive m3 with QoS 1")
+    end,
+    %% Modifying the subscription once more with v3 client should work.
+    {ok, _, [0]} = emqtt:subscribe(C3, Topic, qos0),
+    {ok, _} = emqtt:publish(C3, Topic, <<"m4">>, qos1),
+    receive {publish,
+             #{client_pid := C3,
+               payload := <<"m4">>,
+               qos := 0}} -> ok
+    after 1000 -> ct:fail("did not receive m3 with QoS 0")
+    end,
+
+    %% Unsubscribing in v3 should work.
+    ?assertMatch({ok, _, _}, emqtt:unsubscribe(C3, Topic)),
+    {ok, _} = emqtt:publish(C3, Topic, <<"m5">>, qos1),
+    assert_nothing_received(),
+
+    ok = emqtt:disconnect(C3),
+    C4 = connect(ClientId, Config, Server1, [{proto_ver, v3}, {clean_start, true}]),
+    ok = emqtt:disconnect(C4),
+    eventually(?_assertEqual([], all_connection_pids(Config))).
+
 %% -------------------------------------------------------------------
 %% Helpers
 %% -------------------------------------------------------------------
@@ -555,8 +1217,11 @@ assert_queue_ttl(TTLSecs, NumQs, Config) ->
                       end, Qs)).
 
 dead_letter_metric(Metric, Config) ->
+    dead_letter_metric(Metric, Config, disabled).
+
+dead_letter_metric(Metric, Config, Strategy) ->
     Counters = rpc(Config, rabbit_global_counters, overview, []),
-    Map = maps:get([{queue_type, rabbit_classic_queue}, {dead_letter_strategy, disabled}], Counters),
+    Map = maps:get([{queue_type, rabbit_classic_queue}, {dead_letter_strategy, Strategy}], Counters),
     maps:get(Metric, Map).
 
 assert_nothing_received() ->
diff --git a/moduleindex.yaml b/moduleindex.yaml
index a4880c091d..ced6df0a50 100755
--- a/moduleindex.yaml
+++ b/moduleindex.yaml
@@ -45,6 +45,8 @@ amqp10_common:
 - amqp10_binary_parser
 - amqp10_framing
 - amqp10_framing0
+app:
+- appup_src
 aten:
 - aten
 - aten_app
@@ -55,6 +57,15 @@ aten:
 - aten_sup
 base64url:
 - base64url
+bazdep:
+- bazdep
+certifi:
+- certifi
+- certifi_pt
+codepath:
+- codepath
+cover:
+- foo
 cowboy:
 - cowboy
 - cowboy_app
@@ -111,6 +122,8 @@ credentials_obfuscation:
 ct_helper:
 - ct_helper
 - ct_helper_error_h
+cth_styledout:
+- cth_styledout
 cuttlefish:
 - conf_parse
 - cuttlefish
@@ -135,6 +148,10 @@ cuttlefish:
 - cuttlefish_validator
 - cuttlefish_variable
 - cuttlefish_vmargs
+dummy:
+- dummy_app
+- dummy_server
+- dummy_sup
 eetcd:
 - eetcd_auth_gen
 - eetcd_cluster_gen
@@ -183,10 +200,30 @@ emqtt:
 - emqtt_ws
 enough:
 - enough
+erlc:
+- first_erl
+- foo
+- foo_app
+- foo_test_worker
+- foo_worker
+eunit:
+- foo
+eunit_surefire:
+- foo
+foo:
+- java
+- lisp
+- pascal
+- perl
+foodep:
+- foodep
 gen_batch_server:
 - gen_batch_server
 getopt:
 - getopt
+gpb:
+- gpb
+- gpb_compile
 gun:
 - gun
 - gun_app
@@ -200,6 +237,45 @@ gun:
 - gun_tls
 - gun_ws
 - gun_ws_h
+hackney:
+- hackney
+- hackney_app
+- hackney_bstr
+- hackney_connect
+- hackney_connection
+- hackney_connections
+- hackney_cookie
+- hackney_date
+- hackney_headers
+- hackney_headers_new
+- hackney_http
+- hackney_http_connect
+- hackney_local_tcp
+- hackney_manager
+- hackney_metrics
+- hackney_multipart
+- hackney_pool
+- hackney_pool_handler
+- hackney_request
+- hackney_response
+- hackney_socks5
+- hackney_ssl
+- hackney_ssl_certificate
+- hackney_stream
+- hackney_sup
+- hackney_tcp
+- hackney_trace
+- hackney_url
+- hackney_util
+idna:
+- idna
+- idna_bidi
+- idna_context
+- idna_data
+- idna_mapping
+- idna_table
+- idna_ucs
+- punycode
 inet_tcp_proxy_dist:
 - inet_tcp_proxy_dist
 - inet_tcp_proxy_dist_app
@@ -326,8 +402,18 @@ meck:
 - meck_proc
 - meck_ret_spec
 - meck_util
+metrics:
+- metrics
+- metrics_dummy
+- metrics_exometer
+- metrics_folsom
+mimerl:
+- mimerl
 my_plugin:
 - my_plugin
+neotoma:
+- neotoma
+- neotoma_parse
 observer_cli:
 - observer_cli
 - observer_cli_application
@@ -359,6 +445,13 @@ osiris:
 - osiris_tracking
 - osiris_util
 - osiris_writer
+parse_trans:
+- ct_expand
+- exprecs
+- parse_trans
+- parse_trans_codegen
+- parse_trans_mod
+- parse_trans_pp
 prometheus:
 - prometheus_mnesia_collector
 - prometheus_vm_dist_collector
@@ -417,6 +510,17 @@ proper:
 - proper_unicode
 - proper_unused_imports_remover
 - vararg
+proto_gpb:
+- foo
+- foo_app
+- foo_sup
+proto_protobuffs:
+- foo
+- foo_app
+- foo_sup
+protobuffs:
+- protobuffs
+- protobuffs_compile
 quantile_estimator:
 - quantile
 - quantile_estimator
@@ -1172,6 +1276,51 @@ ranch:
 - ranch_sup
 - ranch_tcp
 - ranch_transport
+rebar:
+- rebar
+- rebar_abnfc_compiler
+- rebar_app_utils
+- rebar_appups
+- rebar_asn1_compiler
+- rebar_base_compiler
+- rebar_cleaner
+- rebar_config
+- rebar_core
+- rebar_cover_utils
+- rebar_ct
+- rebar_deps
+- rebar_dia_compiler
+- rebar_dialyzer
+- rebar_edoc
+- rebar_erlc_compiler
+- rebar_erlydtl_compiler
+- rebar_escripter
+- rebar_eunit
+- rebar_file_utils
+- rebar_getopt
+- rebar_lfe_compiler
+- rebar_log
+- rebar_metacmds
+- rebar_mustache
+- rebar_neotoma_compiler
+- rebar_otp_app
+- rebar_otp_appup
+- rebar_port_compiler
+- rebar_proto_compiler
+- rebar_proto_gpb_compiler
+- rebar_protobuffs_compiler
+- rebar_qc
+- rebar_rand_compat
+- rebar_rel_utils
+- rebar_reltool
+- rebar_require_vsn
+- rebar_shell
+- rebar_subdirs
+- rebar_templater
+- rebar_upgrade
+- rebar_utils
+- rebar_xref
+- rmemo
 recon:
 - recon
 - recon_alloc
@@ -1191,6 +1340,14 @@ seshat:
 - seshat_app
 - seshat_counters_server
 - seshat_sup
+ssl_verify_fun:
+- ssl_verify_fingerprint
+- ssl_verify_fun_cert_helpers
+- ssl_verify_fun_encodings
+- ssl_verify_hostname
+- ssl_verify_pk
+- ssl_verify_string
+- ssl_verify_util
 stdout_formatter:
 - stdout_formatter
 - stdout_formatter_paragraph
@@ -1221,3 +1378,8 @@ trust_store_http:
 - trust_store_http_sup
 - trust_store_invalid_handler
 - trust_store_list_handler
+unicode_util_compat:
+- string_compat
+- unicode_util_compat
+unuseddep:
+- unuseddep