Merge pull request #10244 from rabbitmq/qos0-queue

Allow MQTT QoS 0 subscribers to reconnect

Author: michaelklishin

deps/rabbit/src/rabbit_amqqueue.erl

@@ -1995,7 +1995,8 @@ on_node_down(Node) ->
         {QueueNames, Deletions} ->
             case length(QueueNames) of
                 0 -> ok;
-                _ -> rabbit_log:info("~tp transient queues from an old incarnation of node ~tp deleted in ~fs", [length(QueueNames), Node, Time/1000000])
+                N -> rabbit_log:info("~b transient queues from an old incarnation of node ~tp deleted in ~fs",
+                                     [N, Node, Time / 1_000_000])
             end,
             notify_queue_binding_deletions(Deletions),
             rabbit_core_metrics:queues_deleted(QueueNames),
@@ -2010,6 +2011,7 @@ filter_transient_queues_to_delete(Node) ->
                 andalso (not amqqueue:is_classic(Q) orelse not amqqueue:is_durable(Q))
                 andalso (not rabbit_amqqueue:is_replicated(Q)
                          orelse rabbit_amqqueue:is_dead_exclusive(Q))
+                andalso amqqueue:get_type(Q) =/= rabbit_mqtt_qos0_queue
     end.
 
 notify_queue_binding_deletions(QueueDeletions) when is_list(QueueDeletions) ->

deps/rabbitmq_mqtt/src/rabbit_mqtt_qos0_queue.erl

@@ -68,7 +68,8 @@ is_stateful() ->
     false.
 
 -spec declare(amqqueue:amqqueue(), node()) ->
-    {'new' | 'existing' | 'owner_died', amqqueue:amqqueue()}.
+    {'new' | 'existing' | 'owner_died', amqqueue:amqqueue()} |
+    {'absent', amqqueue:amqqueue(), rabbit_amqqueue:absent_reason()}.
 declare(Q0, _Node) ->
     Q1 = case amqqueue:get_pid(Q0) of
              none ->
@@ -92,20 +93,6 @@ declare(Q0, _Node) ->
                                  {arguments, amqqueue:get_arguments(Q)},
                                  {user_who_performed_action, ActingUser}]),
             {new, Q};
-        {absent, OldQ, nodedown} ->
-            %% This case body can be deleted once Mnesia is unsupported.
-            OldPid = amqqueue:get_pid(OldQ),
-            OldNode = node(OldPid),
-            rabbit_log_queue:debug(
-              "Overwriting record of ~s of type ~s on node ~s since "
-              "formerly hosting node ~s seems to be down (former pid ~p)",
-              [rabbit_misc:rs(amqqueue:get_name(Q1)), ?MODULE, node(), OldNode, OldPid]),
-            case rabbit_amqqueue:internal_declare(Q1, true) of
-                {created, Q} ->
-                    {new, Q};
-                Other ->
-                    Other
-            end;
         Other ->
             Other
     end.

deps/rabbitmq_mqtt/test/shared_SUITE.erl

@@ -1214,38 +1214,45 @@ rabbit_mqtt_qos0_queue(Config) ->
     ok = emqtt:disconnect(Pub).
 
 rabbit_mqtt_qos0_queue_kill_node(Config) ->
-    Topic = atom_to_binary(?FUNCTION_NAME),
+    Topic1 = <<"t/1">>,
+    Topic2 = <<"t/2">>,
     Pub = connect(<<"publisher">>, Config, 2, []),
 
     SubscriberId = <<"subscriber">>,
     Sub0 = connect(SubscriberId, Config, 0, []),
-    {ok, _, [0]} = emqtt:subscribe(Sub0, Topic, qos0),
-    ok = emqtt:publish(Pub, Topic, <<"m0">>, qos0),
-    ok = expect_publishes(Sub0, Topic, [<<"m0">>]),
+    {ok, _, [0]} = emqtt:subscribe(Sub0, Topic1, qos0),
+    ok = emqtt:publish(Pub, Topic1, <<"m0">>, qos0),
+    ok = expect_publishes(Sub0, Topic1, [<<"m0">>]),
 
     process_flag(trap_exit, true),
     ok = rabbit_ct_broker_helpers:kill_node(Config, 0),
-    %% Wait a bit to ensure that Mnesia deletes the queue record on nodes 1 and 2 from Mnesia
-    %% table rabbit_queue (but the queue record is still present in rabbit_durable_queue).
+    ok = await_exit(Sub0),
+    %% Wait to run rabbit_amqqueue:on_node_down/1 on both live nodes.
     timer:sleep(500),
+    %% Re-connect to a live node with same MQTT client ID.
     Sub1 = connect(SubscriberId, Config, 1, []),
-    {ok, _, [0]} = emqtt:subscribe(Sub1, Topic, qos0),
-    ok = emqtt:publish(Pub, Topic, <<"m1">>, qos0),
-    ok = expect_publishes(Sub1, Topic, [<<"m1">>]),
+    {ok, _, [0]} = emqtt:subscribe(Sub1, Topic2, qos0),
+    ok = emqtt:publish(Pub, Topic2, <<"m1">>, qos0),
+    ok = expect_publishes(Sub1, Topic2, [<<"m1">>]),
+    %% Since we started a new clean session, previous subscription should have been deleted.
+    ok = emqtt:publish(Pub, Topic1, <<"m2">>, qos0),
+    receive {publish, _} = Publish -> ct:fail({unexpected, Publish})
+    after 300 -> ok
+    end,
 
-    %% Start node 0 to have a majority for Khepri.
     ok = rabbit_ct_broker_helpers:start_node(Config, 0),
     ok = rabbit_ct_broker_helpers:kill_node(Config, 1),
-    %% This time, do not wait. Mnesia will contain the queue record in rabbit_durable_queue,
-    %% but this time Mnesia may or may not contain the queue record in rabbit_queue.
+    %% This time, do not wait.
+    %% rabbit_amqqueue:on_node_down/1 may or may not have run.
     Sub2 = connect(SubscriberId, Config, 2, []),
-    {ok, _, [0]} = emqtt:subscribe(Sub2, Topic, qos0),
-    ok = emqtt:publish(Pub, Topic, <<"m2">>, qos0),
-    ok = expect_publishes(Sub2, Topic, [<<"m2">>]),
+    {ok, _, [0]} = emqtt:subscribe(Sub2, Topic2, qos0),
+    ok = emqtt:publish(Pub, Topic2, <<"m3">>, qos0),
+    ok = expect_publishes(Sub2, Topic2, [<<"m3">>]),
 
     ok = emqtt:disconnect(Sub2),
     ok = emqtt:disconnect(Pub),
-    ok = rabbit_ct_broker_helpers:start_node(Config, 1).
+    ok = rabbit_ct_broker_helpers:start_node(Config, 1),
+    ?assertEqual([], rpc(Config, rabbit_db_binding, get_all, [])).
 
 %% Test that MQTT connection can be listed and closed via the rabbitmq_management plugin.
 management_plugin_connection(Config) ->