Make sure partitions and binding keys counts are equal

Author: acogoluegnes

deps/rabbitmq_stream/src/rabbit_stream_manager.erl

@@ -229,7 +229,7 @@ handle_call({create_super_stream,
              BindingKeys,
              Username},
             _From, State) ->
-    case validate_super_stream_creation(VirtualHost, Name, Partitions) of
+    case validate_super_stream_creation(VirtualHost, Name, Partitions, BindingKeys) of
         {error, Reason} ->
             {reply, {error, Reason}, State};
         ok ->
@@ -655,7 +655,10 @@ super_stream_partitions(VirtualHost, SuperStream) ->
             {error, stream_not_found}
     end.
 
-validate_super_stream_creation(VirtualHost, Name, Partitions) ->
+validate_super_stream_creation(_VirtualHost, _Name, Partitions, BindingKeys)
+  when length(Partitions) =/= length(BindingKeys) ->
+   {error, {validation_failed, "There must be the same number of partitions and binding keys"}};
+validate_super_stream_creation(VirtualHost, Name, Partitions, _BindingKeys) ->
     case exchange_exists(VirtualHost, Name) of
         {error, validation_failed} ->
             {error,

deps/rabbitmq_stream/test/rabbit_stream_SUITE.erl

@@ -332,7 +332,7 @@ test_super_stream_creation_deletion(Config) ->
     ok = T:send(S, PartitionsFrame),
     {Cmd2, _} = receive_commands(T, S, C),
     ?assertMatch({response, 1, {partitions, ?RESPONSE_CODE_OK, Partitions}},
-                  Cmd2),
+                 Cmd2),
     [begin
          RouteFrame = frame({request, 1, {route, Rk, Ss}}),
          ok = T:send(S, RouteFrame),
@@ -351,7 +351,15 @@ test_super_stream_creation_deletion(Config) ->
     ok = T:send(S, PartitionsFrame),
     {Cmd4, _} = receive_commands(T, S, C),
     ?assertMatch({response, 1, {partitions, ?RESPONSE_CODE_STREAM_DOES_NOT_EXIST, []}},
-                  Cmd4),
+                 Cmd4),
+
+    %% not the same number of partitions and binding keys
+    SsCreationBadFrame = frame({request, 1, {create_super_stream, Ss,
+                                             [<<"s1">>, <<"s2">>], [<<"bk1">>], #{}}}),
+    ok = T:send(S, SsCreationBadFrame),
+    {Cmd5, _} = receive_commands(T, S, C),
+    ?assertMatch({response, 1, {create_super_stream, ?RESPONSE_CODE_PRECONDITION_FAILED}},
+                 Cmd5),
 
     test_close(T, S, C),
     closed = wait_for_socket_close(T, S, 10),

deps/rabbitmq_stream/test/rabbit_stream_manager_SUITE.erl

@@ -99,14 +99,14 @@ lookup_member(Config) ->
     ?assertEqual({ok, deleted}, delete_stream(Config, Stream)).
 
 manage_super_stream(Config) ->
-    % create super stream
+    %% create super stream
     ?assertEqual(ok,
                  create_super_stream(Config,
                                      <<"invoices">>,
                                      [<<"invoices-0">>, <<"invoices-1">>,
                                       <<"invoices-2">>],
                                      [<<"0">>, <<"1">>, <<"2">>])),
-    % get the correct partitions
+    %% get the correct partitions
     ?assertEqual({ok,
                   [<<"invoices-0">>, <<"invoices-1">>, <<"invoices-2">>]},
                  partitions(Config, <<"invoices">>)),
@@ -117,21 +117,21 @@ manage_super_stream(Config) ->
             <- [{<<"invoices-0">>, <<"0">>}, {<<"invoices-1">>, <<"1">>},
                 {<<"invoices-2">>, <<"2">>}]],
 
-    % get an error if trying to re-create it
+    %% get an error if trying to re-create it
     ?assertMatch({error, _},
                  create_super_stream(Config,
                                      <<"invoices">>,
                                      [<<"invoices-0">>, <<"invoices-1">>,
                                       <<"invoices-2">>],
                                      [<<"0">>, <<"1">>, <<"2">>])),
 
-    % can delete it
+    %% can delete it
     ?assertEqual(ok, delete_super_stream(Config, <<"invoices">>)),
 
-    % create a stream with the same name as a potential partition
+    %% create a stream with the same name as a potential partition
     ?assertMatch({ok, _}, create_stream(Config, <<"invoices-1">>)),
 
-    % cannot create the super stream because a partition already exists
+    %% cannot create the super stream because a partition already exists
     ?assertMatch({error, _},
                  create_super_stream(Config,
                                      <<"invoices">>,
@@ -140,6 +140,14 @@ manage_super_stream(Config) ->
                                      [<<"0">>, <<"1">>, <<"2">>])),
 
     ?assertMatch({ok, _}, delete_stream(Config, <<"invoices-1">>)),
+
+    %% not the same number of partitions and binding keys
+    ?assertMatch({error, {validation_failed, _}},
+                 create_super_stream(Config,
+                                     <<"invoices">>,
+                                     [<<"invoices-0">>, <<"invoices-1">>],
+                                     [<<"0">>])),
+
     ok.
 
 partition_index(Config) ->