Stream coordinator: fix member check in ensure_coordinator_started/0

Previously if a command was issued from a rabbit node where there is
not yet a local stream coordinator member it would try to create
a new stream coordinator cluster, which would fail even if there
is a functional cluster spanning the other rabbit nodes.

This was due to use of global:whereis_name to discover stream coordinator
members on other nodes, however, Ra members never register globally
they only do local name registrations.

Replacing with a simple erpc that does erlang:whereis/1

Pairing with @acogoluegnes

2 files changed, 33 insertions, 16 deletions

diff --git a/deps/rabbit/src/rabbit_stream_coordinator.erl b/deps/rabbit/src/rabbit_stream_coordinator.erl
index 87990cdc2f..4371c2de81 100644
--- a/deps/rabbit/src/rabbit_stream_coordinator.erl
+++ b/deps/rabbit/src/rabbit_stream_coordinator.erl
@@ -307,23 +307,25 @@ ensure_coordinator_started() ->
         undefined ->
             global:set_lock(?STREAM_COORDINATOR_STARTUP),
             Nodes = case ra:restart_server(?RA_SYSTEM, Local) of
-                {error, Reason} when Reason == not_started orelse
-                                     Reason == name_not_registered ->
-                    OtherNodes = all_coord_members() -- [Local],
-                    %% We can't use find_members/0 here as a process that timeouts means the cluster is up
-                    case lists:filter(fun(N) -> global:whereis_name(N) =/= undefined end, OtherNodes) of
-                        [] ->
-                            start_coordinator_cluster();
-                        _ ->
-                            OtherNodes
-                    end;
-                ok ->
-                    AllNodes;
-                {error, {already_started, _}} ->
+                        {error, Reason} when Reason == not_started orelse
+                                             Reason == name_not_registered ->
+                            OtherNodes = all_coord_members() -- [Local],
+                            case lists:filter(
+                                   fun({_, N}) ->
+                                           erpc:call(N, erlang, whereis, [?MODULE]) =/= undefined
+                                   end, OtherNodes) of
+                                [] ->
+                                    start_coordinator_cluster();
+                                _ ->
+                                    OtherNodes
+                            end;
+                        ok ->
                             AllNodes;
-                _ ->
-                    AllNodes
-            end,
+                        {error, {already_started, _}} ->
+                            AllNodes;
+                        _ ->
+                            AllNodes
+                    end,
             global:del_lock(?STREAM_COORDINATOR_STARTUP),
             Nodes;
         _ ->
diff --git a/deps/rabbit/test/rabbit_stream_queue_SUITE.erl b/deps/rabbit/test/rabbit_stream_queue_SUITE.erl
index e61996980e..1f197dd655 100644
--- a/deps/rabbit/test/rabbit_stream_queue_SUITE.erl
+++ b/deps/rabbit/test/rabbit_stream_queue_SUITE.erl
@@ -570,7 +570,21 @@ grow_coordinator_cluster(Config) ->
     ok = rabbit_control_helper:command(stop_app, Server1),
     ok = rabbit_control_helper:command(join_cluster, Server1, [atom_to_list(Server0)], []),
     rabbit_control_helper:command(start_app, Server1),
+    %% at this point there _probably_ won't be a stream coordinator member on
+    %% Server1
 
+    %% check we can add a new stream replica for the previously declare stream
+    ?assertEqual(ok,
+                 rpc:call(Server1, rabbit_stream_queue, add_replica,
+                          [<<"/">>, Q, Server1])),
+    %% also check we can declare a new stream when calling Server1
+    Q2 = unicode:characters_to_binary([Q, <<"_2">>]),
+    ChServer1 = rabbit_ct_client_helpers:open_channel(Config, Server1),
+    ?assertEqual({'queue.declare_ok', Q2, 0, 0},
+                 declare(ChServer1, Q2, [{<<"x-queue-type">>, longstr, <<"stream">>}])),
+
+    %% wait until the stream coordinator detects there is a new rabbit node
+    %% and adds a new member on the new node
     rabbit_ct_helpers:await_condition(
       fun() ->
               case rpc:call(Server0, ra, members, [{rabbit_stream_coordinator, Server0}]) of
@@ -589,6 +603,7 @@ shrink_coordinator_cluster(Config) ->
     Ch = rabbit_ct_client_helpers:open_channel(Config, Server0),
     Q = ?config(queue_name, Config),
 
+
     ?assertEqual({'queue.declare_ok', Q, 0, 0},
                  declare(Ch, Q, [{<<"x-queue-type">>, longstr, <<"stream">>}])),