Handle different queue states on queue deletion from the management API more robustly

deps/rabbit/src/amqqueue.erl

@@ -73,6 +73,7 @@
          is_amqqueue/1,
          is_auto_delete/1,
          is_durable/1,
+         is_exclusive/1,
          is_classic/1,
          is_quorum/1,
          pattern_match_all/0,
@@ -557,6 +558,11 @@ is_auto_delete(#amqqueue{auto_delete = AutoDelete}) ->
 
 is_durable(#amqqueue{durable = Durable}) -> Durable.
 
+-spec is_exclusive(amqqueue()) -> boolean().
+
+is_exclusive(Queue) ->
+    is_pid(get_exclusive_owner(Queue)).
+
 -spec is_classic(amqqueue()) -> boolean().
 
 is_classic(Queue) ->

deps/rabbitmq_management/src/rabbit_mgmt_wm_queue.erl

@@ -81,15 +81,13 @@ delete_resource(ReqData, Context = #context{user = #user{username = ActingUser}}
     Name = rabbit_misc:r(VHost, queue, QName),
     case rabbit_amqqueue:lookup(Name) of
         {ok, Q} ->
-            case rabbit_amqqueue:delete(Q, IfUnused, IfEmpty, ActingUser) of
+            IsExclusive = amqqueue:is_exclusive(Q),
+            ExclusiveOwnerPid = amqqueue:get_exclusive_owner(Q),
+            try rabbit_amqqueue:delete_with(Q, ExclusiveOwnerPid, IfUnused, IfEmpty, ActingUser, IsExclusive) of
                 {ok, _} ->
-                    {true, ReqData, Context};
-                {error, not_empty} ->
-                    Explanation = io_lib:format("~ts not empty", [rabbit_misc:rs(Name)]),
-                    rabbit_log:warning("Delete queue error: ~ts", [Explanation]),
-                    rabbit_mgmt_util:bad_request(list_to_binary(Explanation), ReqData, Context);
-                {error, in_use} ->
-                    Explanation = io_lib:format("~ts in use", [rabbit_misc:rs(Name)]),
+                    {true, ReqData, Context}
+            catch
+                _:#amqp_error{explanation = Explanation} ->
                     rabbit_log:warning("Delete queue error: ~ts", [Explanation]),
                     rabbit_mgmt_util:bad_request(list_to_binary(Explanation), ReqData, Context)
             end;

deps/rabbitmq_management/test/rabbit_mgmt_http_SUITE.erl

@@ -83,6 +83,7 @@ all_tests() -> [
     multiple_invalid_connections_test,
     exchanges_test,
     queues_test,
+    crashed_queues_test,
     quorum_queues_test,
     stream_queues_have_consumers_field,
     bindings_test,
@@ -1078,6 +1079,36 @@ queues_test(Config) ->
     http_delete(Config, "/queues/downvhost/bar", {group, '2xx'}),
     passed.
 
+crashed_queues_test(Config) ->
+    Node = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
+    Q = #resource{virtual_host = <<"/">>, kind = queue, name = <<"crashingqueue">>},
+
+    QArgs = #{},
+    http_put(Config, "/queues/%2F/crashingqueue", QArgs, {group, '2xx'}),
+
+    ok = rabbit_ct_broker_helpers:rpc(Config, 0,
+            rabbit_amqqueue_control, await_state, [Node, Q, running]),
+
+    ok = rabbit_ct_broker_helpers:rpc(Config, 0,
+        rabbit_amqqueue, kill_queue_hard, [Node, Q]),
+
+    ok = rabbit_ct_broker_helpers:rpc(Config, 0,
+            rabbit_amqqueue_control, await_state, [Node, Q, crashed]),
+
+    CrashedQueue  = http_get(Config, "/queues/%2F/crashingqueue"),
+
+    assert_item(#{name        => <<"crashingqueue">>,
+                  vhost       => <<"/">>,
+                  state       => <<"crashed">>,
+                  durable     => false,
+                  auto_delete => false,
+                  exclusive   => false,
+                  arguments   => #{}}, CrashedQueue),
+
+    http_delete(Config, "/queues/%2F/crashingqueue", {group, '2xx'}),
+    http_delete(Config, "/queues/%2F/crashingqueue", ?NOT_FOUND),
+    passed.
+
 quorum_queues_test(Config) ->
     %% Test in a loop that no metrics are left behing after deleting a queue
     quorum_queues_test_loop(Config, 5).