rabbit_db_cluster: Check rabbit is stopped in forget_node()

[Why]
`rabbit_mnesia` indirectly checks that `rabbit` is stopped on the remote
node because `mnesia:del_table_copy()` requires that Mnesia is stopped
to delete the schema. However, this is not specific to Mnesia and we
want `rabbit` to be stopped when we use Khepri in the future.

[How]
We use `rabbit:is_running(Node)` to query the status of RabbitMQ on the
remote node to forget. This is not atomic so there is a small chance
that RabbitMQ is restarted between the check and the actual forget.

Note: `rabbit_mnesia` also removes some queues and emit a "left cluster"
event after a successful forget. However, this part was not moved
because other parts of the module rely on this in RPC calls. To keep
nodes compatibles, the calls are left in place. They will be duplicated
for Khepri.

Author: dumbbell

deps/rabbit/src/rabbit_db_cluster.erl

@@ -109,9 +109,17 @@ join_using_mnesia(ClusterNodes, NodeType) when is_list(ClusterNodes) ->
 %% @doc Removes `Node' from the cluster.
 
 forget_member(Node, RemoveWhenOffline) ->
-    rabbit_db:run(
-      #{mnesia =>
-        fun() -> forget_member_using_mnesia(Node, RemoveWhenOffline) end}).
+    case rabbit:is_running(Node) of
+        false ->
+            rabbit_db:run(
+              #{mnesia => fun() ->
+                                  forget_member_using_mnesia(
+                                    Node, RemoveWhenOffline)
+                          end
+               });
+        true ->
+            {error, {failed_to_remove_node, Node, rabbit_still_running}}
+    end.
 
 forget_member_using_mnesia(Node, RemoveWhenOffline) ->
     rabbit_mnesia:forget_cluster_node(Node, RemoveWhenOffline).

deps/rabbitmq_cli/lib/rabbitmq/cli/ctl/commands/forget_cluster_node_command.ex

@@ -58,6 +58,10 @@ defmodule RabbitMQ.CLI.Ctl.Commands.ForgetClusterNodeCommand do
       end
 
     case ret do
+      {:error, {:failed_to_remove_node, ^atom_name, :rabbit_still_running}} ->
+        {:error,
+         "RabbitMQ on node #{node_to_remove} must be stopped with 'rabbitmqctl -n #{node_to_remove} stop_app' before it can be removed"}
+
       {:error, {:failed_to_remove_node, ^atom_name, {:active, _, _}}} ->
         {:error,
          "RabbitMQ on node #{node_to_remove} must be stopped with 'rabbitmqctl -n #{node_to_remove} stop_app' before it can be removed"}