Ensure to only grow QQs when all existing members are in 'voter' status

Author: Ayanda-D

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -1609,19 +1609,43 @@ grow(QuorumClusterSize, _VhostSpec, _QueueSpec, _Strategy, _Membership)
     {error, bad_quorum_cluster_size}.
 
 maybe_grow(Q, Node, Membership, Size) ->
+    QNodes = get_nodes(Q),
+    maybe_grow(Q, Node, Membership, Size, QNodes).
+
+maybe_grow(Q, Node, Membership, Size, QNodes) ->
     QName = amqqueue:get_name(Q),
-    ?LOG_INFO("~ts: adding a new member (replica) on node ~w",
-                    [rabbit_misc:rs(QName), Node]),
-    case add_member(Q, Node, Membership) of
-        ok ->
-            {QName, {ok, Size + 1}};
-        {error, Err} ->
+    {ok, RaName} = qname_to_internal_name(QName),
+    case check_all_memberships(RaName, QNodes, voter) of
+        true ->
+            ?LOG_INFO("~ts: adding a new member (replica) on node ~w",
+                            [rabbit_misc:rs(QName), Node]),
+            case add_member(Q, Node, Membership) of
+                ok ->
+                    {QName, {ok, Size + 1}};
+                {error, Err} ->
+                    ?LOG_WARNING(
+                    "~ts: failed to add member (replica) on node ~w, error: ~w",
+                    [rabbit_misc:rs(QName), Node, Err]),
+                    {QName, {error, Size, Err}}
+            end;
+        false ->
+            Err = {error, non_voters_found},
             ?LOG_WARNING(
-            "~ts: failed to add member (replica) on node ~w, error: ~w",
-            [rabbit_misc:rs(QName), Node, Err]),
+                    "~ts: failed to add member (replica) on node ~w, error: ~w",
+                    [rabbit_misc:rs(QName), Node, Err]),
             {QName, {error, Size, Err}}
     end.
 
+check_all_memberships(RaName, QNodes, CompareMembership) ->
+    case rpc:multicall(QNodes, ets, lookup, [ra_state, RaName]) of
+        {Result, []} ->
+            lists:all(
+                fun(M) -> M == CompareMembership end,
+                [Membership || [{_RaName, _RaState, Membership}] <- Result]);
+        _ ->
+            false
+    end.
+
 -spec transfer_leadership(amqqueue:amqqueue(), node()) -> {migrated, node()} | {not_migrated, atom()}.
 transfer_leadership(Q, Destination) ->
     {RaName, _} = Pid = amqqueue:get_pid(Q),

deps/rabbit/test/quorum_queue_SUITE.erl

@@ -1792,7 +1792,7 @@ dont_leak_file_handles(Config) ->
     ok.
 
 grow_queue(Config) ->
-    [Server0, Server1, _Server2, _Server3, _Server4] =
+    [Server0, Server1, Server2, _Server3, _Server4] =
         rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
 
     Ch = rabbit_ct_client_helpers:open_channel(Config, Server0),
@@ -1825,42 +1825,72 @@ grow_queue(Config) ->
 
     %% grow queues to node 'Server1'
     TargetClusterSize_2 = 2,
-    rpc:call(Server0, rabbit_quorum_queue, grow, [Server1, <<"/">>, <<".*">>, all]),
+    Result1 = rpc:call(Server0, rabbit_quorum_queue, grow, [Server1, <<"/">>, <<".*">>, all]),
+    %% [{{resource,<<"/">>,queue,<<"grow_queue">>},{ok,2}},
+    %%  {{resource,<<"/">>,queue,<<"grow_queue_alt">>},{ok,2}},...]
+    ?assert(lists:all(fun({_, {R, _}}) -> R =:= ok end, Result1)),
     assert_grown_queues(QQs, Server0, TargetClusterSize_2, MsgCount),
 
     %% grow queues to quorum cluster size '2' has no effect
-    rpc:call(Server0, rabbit_quorum_queue, grow, [TargetClusterSize_2, <<"/">>, <<".*">>, all]),
+    Result2 = rpc:call(Server0, rabbit_quorum_queue, grow, [TargetClusterSize_2, <<"/">>, <<".*">>, all]),
+    ?assertEqual([], Result2),
     assert_grown_queues(QQs, Server0, TargetClusterSize_2, MsgCount),
 
     %% grow queues to quorum cluster size '3'
     TargetClusterSize_3 = 3,
-    rpc:call(Server0, rabbit_quorum_queue, grow, [TargetClusterSize_3, <<"/">>, <<".*">>, all]),
+    Result3 = rpc:call(Server0, rabbit_quorum_queue, grow, [TargetClusterSize_3, <<"/">>, <<".*">>, all, voter]),
+    ?assert(lists:all(fun({_, {R, _}}) -> R =:= ok end, Result3)),
     assert_grown_queues(QQs, Server0, TargetClusterSize_3, MsgCount),
 
     %% grow queues to quorum cluster size '5'
     TargetClusterSize_5 = 5,
-    rpc:call(Server0, rabbit_quorum_queue, grow, [TargetClusterSize_5, <<"/">>, <<".*">>, all]),
+    Result4 = rpc:call(Server0, rabbit_quorum_queue, grow, [TargetClusterSize_5, <<"/">>, <<".*">>, all, voter]),
+    ?assert(lists:all(fun({_, {R, _}}) -> R =:= ok end, Result4)),
     assert_grown_queues(QQs, Server0, TargetClusterSize_5, MsgCount),
 
-    %% shrink all queues again
+    %% shrink all queues again down to 1 member
     rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue,
         force_all_queues_shrink_member_to_current_member, []),
     assert_grown_queues(QQs, Server0, TargetClusterSize_1, MsgCount),
 
     %% grow queues to quorum cluster size > '5' (limit = 5).
     TargetClusterSize_10 = 10,
-    rpc:call(Server0, rabbit_quorum_queue, grow, [TargetClusterSize_10, <<"/">>, <<".*">>, all]),
+    Result5 = rpc:call(Server0, rabbit_quorum_queue, grow, [TargetClusterSize_10, <<"/">>, <<".*">>, all]),
+    ?assert(lists:all(fun({_, {R, _}}) -> R =:= ok end, Result5)),
     assert_grown_queues(QQs, Server0, TargetClusterSize_5, MsgCount),
 
-    %% shrink all queues again
+    %% shrink all queues again down to 1 member
     rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue,
         force_all_queues_shrink_member_to_current_member, []),
     assert_grown_queues(QQs, Server0, TargetClusterSize_1, MsgCount),
 
     %% attempt to grow queues to quorum cluster size < '0'.
     BadTargetClusterSize = -5,
     ?assertEqual({error, bad_quorum_cluster_size},
-        rpc:call(Server0, rabbit_quorum_queue, grow, [BadTargetClusterSize, <<"/">>, <<".*">>, all])).
+        rpc:call(Server0, rabbit_quorum_queue, grow, [BadTargetClusterSize, <<"/">>, <<".*">>, all])),
+
+    %% shrink all queues again down to 1 member
+    rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue,
+        force_all_queues_shrink_member_to_current_member, []),
+    assert_grown_queues(QQs, Server0, TargetClusterSize_1, MsgCount),
+
+    %% grow queues to node 'Server1': non_voter
+    rpc:call(Server0, rabbit_quorum_queue, grow, [Server1, <<"/">>, <<".*">>, all, non_voter]),
+    assert_grown_queues(QQs, Server0, TargetClusterSize_2, MsgCount),
+
+    %% grow queues to node 'Server2': fail, non_voters found
+    Result6 = rpc:call(Server0, rabbit_quorum_queue, grow, [Server2, <<"/">>, <<".*">>, all, voter]),
+    %% [{{resource,<<"/">>,queue,<<"grow_queue">>},{error, 2, {error, non_voters_found}},
+    %%  {{resource,<<"/">>,queue,<<"grow_queue_alt">>},{error, 2, {error, non_voters_found}},...]
+    ?assert(lists:all(
+        fun({_, Err}) -> Err =:= {error, TargetClusterSize_2, {error, non_voters_found}} end, Result6)),
+    assert_grown_queues(QQs, Server0, TargetClusterSize_2, MsgCount),
+
+    %% grow queues to target quorum cluster size '5': fail, non_voters found
+    Result7 = rpc:call(Server0, rabbit_quorum_queue, grow, [TargetClusterSize_5, <<"/">>, <<".*">>, all]),
+    ?assert(lists:all(
+        fun({_, Err}) -> Err =:= {error, TargetClusterSize_2, {error, non_voters_found}} end, Result7)),
+    assert_grown_queues(QQs, Server0, TargetClusterSize_2, MsgCount).
 
 assert_grown_queues(Qs, Node, TargetClusterSize, MsgCount) ->
     [begin

deps/rabbitmq_cli/lib/rabbitmq/cli/queues/commands/grow_command.ex

@@ -44,6 +44,11 @@ defmodule RabbitMQ.CLI.Queues.Commands.GrowCommand do
     {:validation_failure, "target quorum cluster size '#{n}' must be greater than 0."}
   end
 
+  def validate([n, _], %{membership: m})
+      when (is_integer(n) and not (m == "voter" or m == "promotable")) do
+    {:validation_failure, "voter status '#{m}' must be 'voter' or 'promotable' to grow to target quorum cluster size '#{n}'."}
+  end
+
   def validate(_, %{membership: m})
       when not (m == "promotable" or
                   m == "non_voter" or

deps/rabbitmq_cli/test/queues/grow_command_test.exs

@@ -82,10 +82,14 @@ defmodule RabbitMQ.CLI.Queues.Commands.GrowCommandTest do
              {:validation_failure, "voter status 'banana' is not recognised."}
   end
 
-  test "validate: when target quorum cluster size greater than zero, returns a success" do
+  test "validate: when target quorum cluster size greater than zero and membership is voter, returns a success" do
     assert @command.validate([7, "all"], %{membership: "voter", queue_pattern: "qq.*"}) == :ok
   end
 
+  test "validate: when target quorum cluster size greater than zero and membership is promotable, returns a success" do
+    assert @command.validate([5, "all"], %{membership: "promotable", queue_pattern: "qq.*"}) == :ok
+  end
+
   test "validate: when target quorum cluster size is zero, returns failure" do
     assert @command.validate([0, "all"], %{membership: "voter", queue_pattern: "qq.*"}) ==
              {:validation_failure, "target quorum cluster size '0' must be greater than 0."}
@@ -96,6 +100,11 @@ defmodule RabbitMQ.CLI.Queues.Commands.GrowCommandTest do
              {:validation_failure, "target quorum cluster size '-1' must be greater than 0."}
   end
 
+  test "validate: when target quorum cluster size is provided and membership is not voter, returns failure" do
+    assert @command.validate([5, "all"], %{membership: "non_voter", queue_pattern: "qq.*"}) ==
+             {:validation_failure, "voter status 'non_voter' must be 'voter' or 'promotable' to grow to target quorum cluster size '5'."}
+  end
+
   @tag test_timeout: 3000
   test "run: targeting an unreachable node throws a badrpc when growing to a target node", context do
     assert match?(