Merge pull request #10205 from rabbitmq/mergify/bp/v3.12.x/pr-10203

Overwrite rabbit_mqtt_qos0_queue record from crashed node (backport #10203)

Author: michaelklishin

deps/rabbitmq_mqtt/src/rabbit_mqtt_qos0_queue.erl

@@ -67,8 +67,7 @@ is_stateful() ->
     false.
 
 -spec declare(amqqueue:amqqueue(), node()) ->
-    {'new' | 'existing' | 'owner_died', amqqueue:amqqueue()} |
-    {'absent', amqqueue:amqqueue(), rabbit_amqqueue:absent_reason()}.
+    {'new' | 'existing' | 'owner_died', amqqueue:amqqueue()}.
 declare(Q0, _Node) ->
     %% The queue gets persisted such that routing to this
     %% queue (via the topic exchange) works as usual.
@@ -85,6 +84,20 @@ declare(Q0, _Node) ->
                                  {arguments, amqqueue:get_arguments(Q0)},
                                  {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(Q0)), ?MODULE, node(), OldNode, OldPid]),
+            case rabbit_amqqueue:internal_declare(Q0, true) of
+                {created, Q} ->
+                    {new, Q};
+                Other ->
+                    Other
+            end;
         Other ->
             Other
     end.

deps/rabbitmq_mqtt/test/shared_SUITE.erl

@@ -109,6 +109,7 @@ subgroups() ->
        flow_quorum_queue,
        flow_stream,
        rabbit_mqtt_qos0_queue,
+       rabbit_mqtt_qos0_queue_kill_node,
        cli_list_queues,
        maintenance,
        delete_create_queue,
@@ -1019,6 +1020,40 @@ rabbit_mqtt_qos0_queue(Config) ->
     ok = emqtt:disconnect(Sub),
     ok = emqtt:disconnect(Pub).
 
+rabbit_mqtt_qos0_queue_kill_node(Config) ->
+    Topic = atom_to_binary(?FUNCTION_NAME),
+    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">>]),
+
+    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).
+    timer:sleep(500),
+    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">>]),
+
+    %% 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.
+    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 = emqtt:disconnect(Sub2),
+    ok = emqtt:disconnect(Pub),
+    ok = rabbit_ct_broker_helpers:start_node(Config, 1).
+
 %% Test that MQTT connection can be listed and closed via the rabbitmq_management plugin.
 management_plugin_connection(Config) ->
     KeepaliveSecs = 99,