diff options
| author | Michael Klishin <klishinm@vmware.com> | 2022-10-19 20:46:07 +0400 |
|---|---|---|
| committer | GitHub <noreply@github.com> | 2022-10-19 20:46:07 +0400 |
| commit | 666656d996d5801a5984c2315b50778f8a5c09b6 (patch) | |
| tree | 2863e9932a6e439726700b0d98138c42bfad3c78 | |
| parent | 1f6ead8055297dd5b798a60a00bf2e647753e693 (diff) | |
| parent | 0cecb97a2931b4e20768e6219a5ff9b141d3c5e9 (diff) | |
| download | rabbitmq-server-git-666656d996d5801a5984c2315b50778f8a5c09b6.tar.gz | |
Merge pull request #6182 from rabbitmq/stream-coordinator-cluster-creation-fix
Stream coordinator: fix member check in ensure_coordinator_started/0
| -rw-r--r-- | deps/rabbit/src/rabbit_stream_coordinator.erl | 34 | ||||
| -rw-r--r-- | deps/rabbit/test/rabbit_stream_queue_SUITE.erl | 15 |
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">>}])), |