Add initial-cluster-size as a stream policystream-policies

It was only a queue argument

3 files changed, 37 insertions, 8 deletions

diff --git a/src/rabbit_policies.erl b/src/rabbit_policies.erl
index f2c44e2289..54e4d2c03e 100644
--- a/src/rabbit_policies.erl
+++ b/src/rabbit_policies.erl
@@ -42,6 +42,7 @@ register() ->
                           {policy_validator, <<"max-age">>},
                           {policy_validator, <<"max-segment-size">>},
                           {policy_validator, <<"queue-leader-locator">>},
+                          {policy_validator, <<"initial-cluster-size">>},
                           {operator_policy_validator, <<"expires">>},
                           {operator_policy_validator, <<"message-ttl">>},
                           {operator_policy_validator, <<"max-length">>},
@@ -157,6 +158,12 @@ validate_policy0(<<"queue-leader-locator">>, <<"least-leaders">>) ->
 validate_policy0(<<"queue-leader-locator">>, Value) ->
     {error, "~p is not a valid queue leader locator value", [Value]};
 
+validate_policy0(<<"initial-cluster-size">>, Value)
+  when is_integer(Value), Value >= 0 ->
+    ok;
+validate_policy0(<<"initial-cluster-size">>, Value) ->
+    {error, "~p is not a valid cluster size", [Value]};
+
 validate_policy0(<<"max-segment-size">>, Value)
   when is_integer(Value), Value >= 0 ->
     ok;
diff --git a/src/rabbit_stream_queue.erl b/src/rabbit_stream_queue.erl
index 0ae58fc6b8..8f9a51a17d 100644
--- a/src/rabbit_stream_queue.erl
+++ b/src/rabbit_stream_queue.erl
@@ -506,9 +506,11 @@ make_stream_conf(Node, Q) ->
     MaxSegmentSize = args_policy_lookup(<<"max-segment-size">>, fun min/2, Q),
     LeaderLocator = queue_leader_locator(args_policy_lookup(<<"queue-leader-locator">>,
                                                             fun res_arg/2, Q)),
+    InitialClusterSize = initial_cluster_size(args_policy_lookup(<<"initial-cluster-size">>,
+                                                                 fun res_arg/2, Q)),
     Replicas0 = rabbit_mnesia:cluster_nodes(all) -- [Node],
     Arguments = amqqueue:get_arguments(Q),
-    Replicas = select_stream_nodes(get_initial_cluster_size(Arguments) - 1, Replicas0),
+    Replicas = select_stream_nodes(InitialClusterSize - 1, Replicas0),
     Formatter = {?MODULE, format_osiris_event, [QName]},
     Retention = lists:filter(fun({_, R}) ->
                                      R =/= undefined
@@ -576,6 +578,11 @@ max_age(Age1, Age2) ->
 queue_leader_locator(undefined) -> <<"client-local">>;
 queue_leader_locator(Val) -> Val.
 
+initial_cluster_size(undefined) ->
+    length(rabbit_mnesia:cluster_nodes(running));
+initial_cluster_size(Val) ->
+    Val.
+
 res_arg(PolVal, undefined) -> PolVal;
 res_arg(_, ArgVal) ->  ArgVal.
     
@@ -688,7 +695,7 @@ msg_to_iodata(#basic_message{exchange_name = #resource{name = Exchange},
 
 capabilities() ->
     #{policies => [<<"max-length-bytes">>, <<"max-age">>, <<"max-segment-size">>,
-                   <<"queue-leader-locator">>],
+                   <<"queue-leader-locator">>, <<"initial-cluster-size">>],
       queue_arguments => [<<"x-dead-letter-exchange">>, <<"x-dead-letter-routing-key">>,
                           <<"x-max-length">>, <<"x-max-length-bytes">>,
                           <<"x-single-active-consumer">>, <<"x-queue-type">>,
@@ -696,9 +703,3 @@ capabilities() ->
                           <<"x-initial-cluster-size">>, <<"x-queue-leader-locator">>],
       consumer_arguments => [<<"x-stream-offset">>],
       server_named => false}.
-
-get_initial_cluster_size(Arguments) ->
-    case rabbit_misc:table_lookup(Arguments, <<"x-initial-cluster-size">>) of
-        undefined -> length(rabbit_mnesia:cluster_nodes(running));
-        {_Type, Val} -> Val
-    end.
diff --git a/test/rabbit_stream_queue_SUITE.erl b/test/rabbit_stream_queue_SUITE.erl
index 5ae4de82f4..06f69569e8 100644
--- a/test/rabbit_stream_queue_SUITE.erl
+++ b/test/rabbit_stream_queue_SUITE.erl
@@ -49,6 +49,7 @@ groups() ->
            leader_failover,
            initial_cluster_size_one,
            initial_cluster_size_two,
+           initial_cluster_size_one_policy,
            leader_locator_client_local,
            leader_locator_random,
            leader_locator_least_leaders,
@@ -1144,6 +1145,26 @@ initial_cluster_size_two(Config) ->
     ?assertMatch(#'queue.delete_ok'{},
                  amqp_channel:call(Ch, #'queue.delete'{queue = Q})).
 
+initial_cluster_size_one_policy(Config) ->
+    [Server1 | _] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+
+    ok = rabbit_ct_broker_helpers:set_policy(
+           Config, 0, <<"cluster-size">>, <<"initial_cluster_size_one_policy">>, <<"queues">>,
+           [{<<"initial-cluster-size">>, 1}]),
+
+    Ch = rabbit_ct_client_helpers:open_channel(Config, Server1),
+    Q = ?config(queue_name, Config),
+
+    ?assertEqual({'queue.declare_ok', Q, 0, 0},
+                 declare(Ch, Q, [{<<"x-queue-type">>, longstr, <<"stream">>},
+                                  {<<"x-initial-cluster-size">>, long, 1}])),
+    check_leader_and_replicas(Config, Q, Server1, []),
+
+    ?assertMatch(#'queue.delete_ok'{},
+                 amqp_channel:call(Ch, #'queue.delete'{queue = Q})),
+
+    ok = rabbit_ct_broker_helpers:clear_policy(Config, 0, <<"cluster-size">>).
+
 leader_locator_client_local(Config) ->
     [Server1, Server2, Server3] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),