Merge pull request #16710 from rabbitmq/mergify/bp/v4.3.x/pr-16706

Handle infinity for max super stream partitions (backport #16706)

Author: michaelklishin

deps/rabbitmq_stream/src/Elixir.RabbitMQ.CLI.Ctl.Commands.AddSuperStreamCommand.erl

@@ -176,8 +176,8 @@ run([SuperStream],
       timeout := Timeout,
       partitions := Partitions} =
         Opts) ->
-    Streams = streams_from_partitions(SuperStream, Partitions),
-    RoutingKeys = routing_keys(Partitions),
+    Streams = rabbit_stream_utils:streams_from_partitions(SuperStream, Partitions),
+    RoutingKeys = rabbit_stream_utils:routing_keys(Partitions),
     create_super_stream(NodeName,
                         Timeout,
                         VHost,
@@ -191,8 +191,9 @@ run([SuperStream],
       timeout := Timeout,
       binding_keys := BindingKeysStr} =
         Opts) ->
-    BindingKeys = binding_keys(BindingKeysStr),
-    Streams = streams_from_binding_keys(SuperStream, BindingKeys),
+    BindingKeys = rabbit_stream_utils:binding_keys(BindingKeysStr),
+    Streams = rabbit_stream_utils:streams_from_binding_keys(SuperStream,
+                                                            BindingKeys),
 
     create_super_stream(NodeName,
                         Timeout,
@@ -202,24 +203,6 @@ run([SuperStream],
                         stream_arguments(Opts),
                         BindingKeys).
 
-streams_from_partitions(Name, Partitions) ->
-    [<<Name/binary, "-", (integer_to_binary(K))/binary>> ||
-     K <- lists:seq(0, Partitions - 1)].
-
-routing_keys(Partitions) ->
-    [integer_to_binary(K) || K <- lists:seq(0, Partitions - 1)].
-
-binding_keys(BindingKeysBin) ->
-    Keys = binary:split(rabbit_data_coercion:to_binary(BindingKeysBin),
-                        <<",">>, [global]),
-    %% Trim first, then verify the token is not an empty binary
-    [Trimmed || K <- Keys,
-                Trimmed <- [string:trim(K)],
-                Trimmed =/= <<>>].
-
-streams_from_binding_keys(Name, BindingKeys) ->
-    [<<Name/binary, "-", K/binary>> || K <- BindingKeys].
-
 stream_arguments(Opts) ->
     stream_arguments(#{}, Opts).
 

deps/rabbitmq_stream/src/rabbit_stream_manager.erl

@@ -674,14 +674,14 @@ validate_super_stream_partitions(Partitions) ->
     end.
 
 validate_super_stream_max_partitions(Partitions) ->
-    MaxPartitions = rabbit_stream_utils:max_super_stream_partitions(),
-    case erlang:length(Partitions) > MaxPartitions of
+    case rabbit_stream_utils:validate_super_stream_max_partitions(Partitions) of
         true ->
+            ok;
+        false ->
+            MaxPartitions = rabbit_stream_utils:max_super_stream_partitions(),
             {error, {validation_failed,
                      {rabbit_misc:format("The partition number must not exceed ~w",
-                                         [MaxPartitions])}}};
-        false ->
-            ok
+                                         [MaxPartitions])}}}
     end.
 
 exchange_exists(VirtualHost, Name) ->

deps/rabbitmq_stream/src/rabbit_stream_utils.erl

@@ -40,8 +40,18 @@
          check_super_stream_permitted/3,
          offset_lag/4,
          consumer_offset/3,
+         validate_super_stream_max_partitions/1,
          max_super_stream_partitions/0]).
 
+%% super stream partition helpers
+-export([streams_from_partitions/2,
+         streams_from_binding_keys/2,
+         binding_keys/1,
+         routing_keys/1]).
+
+%% for tests
+-export([validate_super_stream_max_partitions/2]).
+
 -include_lib("rabbit_common/include/rabbit.hrl").
 -include_lib("rabbitmq_stream_common/include/rabbit_stream.hrl").
 -include_lib("kernel/include/logger.hrl").
@@ -361,6 +371,44 @@ offset_lag(_, 0, 0, LastListenerOffset) when LastListenerOffset > 0 ->
 offset_lag(CommittedOffset, ConsumerOffset, _, _) ->
     CommittedOffset - ConsumerOffset.
 
+-spec validate_super_stream_max_partitions(list() | integer()) -> boolean().
+validate_super_stream_max_partitions(Partitions) ->
+    MaxPartitions = rabbit_stream_utils:max_super_stream_partitions(),
+    validate_super_stream_max_partitions(Partitions, MaxPartitions).
+
+-spec validate_super_stream_max_partitions(list() | integer(),
+                                           infinity | non_neg_integer()) -> boolean().
+validate_super_stream_max_partitions(_, infinity) ->
+    true;
+validate_super_stream_max_partitions(Partitions, Max) when is_list(Partitions) ->
+    length(Partitions) =< Max;
+validate_super_stream_max_partitions(Partitions, Max) when is_integer(Partitions) ->
+    Partitions =< Max.
+
+-spec max_super_stream_partitions() -> infinity | non_neg_integer().
 max_super_stream_partitions() ->
     application:get_env(rabbitmq_stream, max_super_stream_partitions,
                         ?MAX_SUPER_STREAM_PARTITIONS).
+
+%% super stream partition helpers
+-spec streams_from_partitions(binary(), non_neg_integer()) -> [binary()].
+streams_from_partitions(Name, Partitions) ->
+    [<<Name/binary, "-", (integer_to_binary(K))/binary>> ||
+     K <- lists:seq(0, Partitions - 1)].
+
+-spec streams_from_binding_keys(binary(), [binary()]) -> [binary()].
+streams_from_binding_keys(Name, BindingKeys) ->
+    [<<Name/binary, "-", K/binary>> || K <- BindingKeys].
+
+-spec routing_keys(non_neg_integer()) -> [binary()].
+routing_keys(Partitions) ->
+    [integer_to_binary(K) || K <- lists:seq(0, Partitions - 1)].
+
+-spec binding_keys(unicode:chardata()) -> [binary()].
+binding_keys(BindingKeysBin) ->
+    Keys = binary:split(rabbit_data_coercion:to_binary(BindingKeysBin),
+                        <<",">>, [global]),
+    %% Trim first, then verify the token is not an empty binary
+    [Trimmed || K <- Keys,
+                Trimmed <- [string:trim(K)],
+                Trimmed =/= <<>>].

deps/rabbitmq_stream/test/rabbit_stream_manager_SUITE.erl

@@ -18,13 +18,16 @@
 
 -compile(export_all).
 
+-define(TEST_MAX_SUPER_STREAM_PARTITIONS, 5).
+
 all() ->
     [{group, non_parallel_tests}].
 
 groups() ->
     [{non_parallel_tests, [],
       [manage_super_stream,
        manage_super_stream_max_partitions,
+       manage_super_stream_max_partitions_infinity,
        lookup_leader,
        lookup_member,
        partition_index,
@@ -58,7 +61,7 @@ init_per_group(_, Config) ->
                     1000}]},
     RmqStreamAppConf = {rabbitmq_stream,
                         [{connection_negotiation_step_timeout, 500},
-                         {max_super_stream_partitions, 5}]},
+                         {max_super_stream_partitions, ?TEST_MAX_SUPER_STREAM_PARTITIONS}]},
     run_setup_steps(Config1,
                     [fun(StepConfig) ->
                              merge_app_env(StepConfig, RmqAppConf)
@@ -72,9 +75,24 @@ end_per_group(_, Config) ->
     rabbit_ct_helpers:run_steps(Config,
                                 rabbit_ct_broker_helpers:teardown_steps()).
 
+init_per_testcase(manage_super_stream_max_partitions_infinity = TestCase, Config) ->
+    ok = rabbit_ct_broker_helpers:rpc(Config,
+                                      0,
+                                      application,
+                                      set_env,
+                                      [rabbitmq_stream, max_super_stream_partitions, infinity]),
+    rabbit_ct_helpers:testcase_started(Config, TestCase);
 init_per_testcase(Testcase, Config) ->
     rabbit_ct_helpers:testcase_started(Config, Testcase).
 
+end_per_testcase(manage_super_stream_max_partitions_infinity = TestCase, Config) ->
+    ok = rabbit_ct_broker_helpers:rpc(Config,
+                                      0,
+                                      application,
+                                      set_env,
+                                      [rabbitmq_stream, max_super_stream_partitions,
+                                       ?TEST_MAX_SUPER_STREAM_PARTITIONS]),
+    rabbit_ct_helpers:testcase_finished(Config, TestCase);
 end_per_testcase(Testcase, Config) ->
     rabbit_ct_helpers:testcase_finished(Config, Testcase).
 
@@ -157,7 +175,7 @@ manage_super_stream(Config) ->
     ok.
 
 manage_super_stream_max_partitions(Config) ->
-    Partitions = 6,
+    Partitions = ?TEST_MAX_SUPER_STREAM_PARTITIONS + 1,
     Name = <<"invoices">>,
     Streams =[<<Name/binary, "-", (integer_to_binary(K))/binary>> ||
               K <- lists:seq(0, Partitions - 1)],
@@ -167,6 +185,22 @@ manage_super_stream_max_partitions(Config) ->
                   {validation_failed,
                    {"The partition number must not exceed 5"}}},
                  create_super_stream(Config, Name, Streams, RKs)),
+
+    ok.
+
+manage_super_stream_max_partitions_infinity(Config) ->
+    Partitions = ?TEST_MAX_SUPER_STREAM_PARTITIONS + 1,
+    Name = <<"invoices">>,
+    Streams =[<<Name/binary, "-", (integer_to_binary(K))/binary>> ||
+              K <- lists:seq(0, Partitions - 1)],
+    RKs = [integer_to_binary(K) || K <- lists:seq(0, Partitions - 1)],
+
+    ?assertEqual(ok,
+                 create_super_stream(Config, Name, Streams, RKs)),
+    ?assertEqual({ok, Streams},
+                 partitions(Config, <<"invoices">>)),
+
+    ?assertEqual(ok, delete_super_stream(Config, <<"invoices">>)),
     ok.
 
 partition_index(Config) ->

deps/rabbitmq_stream/test/rabbit_stream_utils_SUITE.erl

@@ -6,6 +6,9 @@
 -include_lib("eunit/include/eunit.hrl").
 -include_lib("rabbit_common/include/rabbit.hrl").
 
+-import(rabbit_stream_utils,
+        [validate_super_stream_max_partitions/2]).
+
 %%%===================================================================
 %%% Common Test callbacks
 %%%===================================================================
@@ -17,7 +20,12 @@ suite() ->
     [{timetrap, {seconds, 30}}].
 
 groups() ->
-    [{tests, [], [sort_partitions, filter_spec, filter_defined]}].
+    [{tests, [],
+      [sort_partitions,
+       filter_spec,
+       filter_defined,
+       test_validate_max_super_stream_partitions,
+       super_stream_partition_helpers]}].
 
 init_per_suite(Config) ->
     Config.
@@ -94,6 +102,66 @@ filter_spec(_) ->
     #{} = rabbit_stream_utils:filter_spec(#{<<"sac">> => true}),
     ok.
 
+test_validate_max_super_stream_partitions(_) ->
+    %% infinity means no limit applies
+    ?assertEqual(true, validate_super_stream_max_partitions([], infinity)),
+    ?assertEqual(true, validate_super_stream_max_partitions([a, b, c], infinity)),
+    ?assertEqual(true, validate_super_stream_max_partitions(0, infinity)),
+    ?assertEqual(true, validate_super_stream_max_partitions(1000, infinity)),
+    %% max = 0: only an empty list and integer 0 are valid
+    ?assertEqual(true, validate_super_stream_max_partitions([], 0)),
+    ?assertEqual(false, validate_super_stream_max_partitions([a], 0)),
+    ?assertEqual(true, validate_super_stream_max_partitions(0, 0)),
+    ?assertEqual(false, validate_super_stream_max_partitions(1, 0)),
+    %% exactly at the limit is valid
+    ?assertEqual(true, validate_super_stream_max_partitions([a, b, c], 3)),
+    ?assertEqual(true, validate_super_stream_max_partitions(3, 3)),
+    %% one over the limit is invalid
+    ?assertEqual(false, validate_super_stream_max_partitions([a, b, c, d], 3)),
+    ?assertEqual(false, validate_super_stream_max_partitions(4, 3)),
+    ok.
+
+super_stream_partition_helpers(_) ->
+    %% streams_from_partitions/2
+    ?assertEqual([],
+                 rabbit_stream_utils:streams_from_partitions(<<"invoices">>, 0)),
+    ?assertEqual([<<"invoices-0">>],
+                 rabbit_stream_utils:streams_from_partitions(<<"invoices">>, 1)),
+    ?assertEqual([<<"invoices-0">>, <<"invoices-1">>, <<"invoices-2">>],
+                 rabbit_stream_utils:streams_from_partitions(<<"invoices">>, 3)),
+
+    %% streams_from_binding_keys/2
+    ?assertEqual([],
+                 rabbit_stream_utils:streams_from_binding_keys(<<"invoices">>, [])),
+    ?assertEqual([<<"invoices-amer">>],
+                 rabbit_stream_utils:streams_from_binding_keys(<<"invoices">>,
+                                                               [<<"amer">>])),
+    ?assertEqual([<<"invoices-amer">>, <<"invoices-emea">>, <<"invoices-apac">>],
+                 rabbit_stream_utils:streams_from_binding_keys(<<"invoices">>,
+                                                               [<<"amer">>,
+                                                                <<"emea">>,
+                                                                <<"apac">>])),
+
+    %% routing_keys/1
+    ?assertEqual([], rabbit_stream_utils:routing_keys(0)),
+    ?assertEqual([<<"0">>], rabbit_stream_utils:routing_keys(1)),
+    ?assertEqual([<<"0">>, <<"1">>, <<"2">>],
+                 rabbit_stream_utils:routing_keys(3)),
+
+    %% binding_keys/1
+    ?assertEqual([<<"amer">>],
+                 rabbit_stream_utils:binding_keys(<<"amer">>)),
+    ?assertEqual([<<"amer">>, <<"emea">>, <<"apac">>],
+                 rabbit_stream_utils:binding_keys(<<"amer,emea,apac">>)),
+    ?assertEqual([<<"amer">>, <<"emea">>, <<"apac">>],
+                 rabbit_stream_utils:binding_keys(<<"amer, emea, apac">>)),
+    ?assertEqual([],
+                 rabbit_stream_utils:binding_keys(<<"">>)),
+    ?assertEqual([],
+                 rabbit_stream_utils:binding_keys(<<"  ">>)),
+
+    ok.
+
 binding(Destination, Order) ->
     #binding{destination = #resource{name = Destination},
              args = [{<<"x-stream-partition-order">>, signedint, Order}]}.