Add partition integration test for super stream SAC group

Author: acogoluegnes

deps/rabbitmq_stream/test/rabbit_stream_partitions_SUITE.erl

@@ -11,7 +11,7 @@
 %% The Original Code is RabbitMQ.
 %%
 %% The Initial Developer of the Original Code is Pivotal Software, Inc.
-%% Copyright (c) 2020-2025 Broadcom. All Rights Reserved.
+%% Copyright (c) 2025 Broadcom. All Rights Reserved.
 %% The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
 %%
 
@@ -38,6 +38,7 @@ groups() ->
     [{cluster, [],
       [simple_sac_consumer_should_get_disconnected_on_network_partition,
        simple_sac_consumer_should_get_disconnected_on_coord_leader_network_partition,
+       super_stream_sac_consumer_should_get_disconnected_on_network_partition,
        super_stream_sac_consumer_should_get_disconnected_on_coord_leader_network_partition]}
     ].
 
@@ -94,13 +95,22 @@ end_per_testcase(TestCase, Config) ->
 
 
 simple_sac_consumer_should_get_disconnected_on_network_partition(Config) ->
+    init_coordinator(Config),
+    CL = coordinator_leader(Config),
+
     S = rabbit_data_coercion:to_binary(?FUNCTION_NAME),
+    init_stream(Config, CL, S),
+
+    [L, F1, F2] = topology(Config, S),
+
+    %% the stream leader and the coordinator leader are on the same node
+    %% another node will be isolated
+    ?assertEqual(L#node.name, coordinator_leader(Config)),
+
     {ok, So0, C0_00} = stream_test_utils:connect(Config, 0),
     {ok, So1, C1_00} = stream_test_utils:connect(Config, 1),
     {ok, So2, C2_00} = stream_test_utils:connect(Config, 2),
 
-    init_stream(Config, 0, S),
-
     C0_01 = register_sac(So0, C0_00, S, 0),
     C0_02 = receive_consumer_update(So0, C0_01),
 
@@ -110,33 +120,33 @@ simple_sac_consumer_should_get_disconnected_on_network_partition(Config) ->
                       1 => {So1, C1_01},
                       2 => {So2, C2_01}},
 
-    Members = stream_members(Config, S),
-    L = leader(Members),
-    [F1, F2] = followers(Members),
-
     Consumers1 = query_consumers(Config, S),
     assertSize(3, Consumers1),
     assertConsumersConnected(Consumers1),
 
-    Isolated = F1,
+    LN = L#node.name,
+    F1N = F1#node.name,
+    F2N = F2#node.name,
+
+    Isolated = F1N,
     {value, DisconnectedConsumer} =
         lists:search(fun(#consumer{pid = ConnPid}) ->
                              rpc(Config, erlang, node, [ConnPid]) =:= Isolated
                      end, Consumers1),
     #consumer{subscription_id = DiscSubId} = DisconnectedConsumer,
 
-    rabbit_ct_broker_helpers:block_traffic_between(Isolated, L),
-    rabbit_ct_broker_helpers:block_traffic_between(Isolated, F2),
+    rabbit_ct_broker_helpers:block_traffic_between(Isolated, LN),
+    rabbit_ct_broker_helpers:block_traffic_between(Isolated, F2N),
 
-    wait_for_disconnected_consumer(Config, L, S),
-    wait_for_forgotten_consumer(Config, L, S),
+    wait_for_disconnected_consumer(Config, LN, S),
+    wait_for_forgotten_consumer(Config, LN, S),
 
-    rabbit_ct_broker_helpers:allow_traffic_between(Isolated, L),
-    rabbit_ct_broker_helpers:allow_traffic_between(Isolated, F2),
+    rabbit_ct_broker_helpers:allow_traffic_between(Isolated, LN),
+    rabbit_ct_broker_helpers:allow_traffic_between(Isolated, F2N),
 
-    wait_for_all_consumers_connected(Config, L, S),
+    wait_for_all_consumers_connected(Config, LN, S),
 
-    Consumers2 = query_consumers(Config, L, S),
+    Consumers2 = query_consumers(Config, LN, S),
     %% the disconnected, then forgotten consumer is cancelled,
     %% because the stream member on its node has been restarted
     assertSize(2, Consumers2),
@@ -168,20 +178,21 @@ simple_sac_consumer_should_get_disconnected_on_network_partition(Config) ->
     ok.
 
 simple_sac_consumer_should_get_disconnected_on_coord_leader_network_partition(Config) ->
-    %% to initialize the stream coordinator on the first node
-    init_stream(Config, 0, <<"dummy">>),
-    delete_stream(stream_port(Config, 0), <<"dummy">>),
+    init_coordinator(Config),
+    CL = coordinator_leader(Config),
+    [CF1, CF2] = all_nodes(Config) -- [CL],
 
     S = rabbit_data_coercion:to_binary(?FUNCTION_NAME),
-
-    init_stream(Config, 1, S),
+    init_stream(Config, CF1, S),
     [L, _F1, _F2] = topology(Config, S),
 
-    wait_for_coordinator_ready(Config),
+    %% the stream leader and the coordinator leader are not on the same node
+    %% the coordinator leader node will be isolated
+    ?assertNotEqual(L#node.name, CL),
 
-    {ok, So0, C0_00} = stream_test_utils:connect(Config, 0),
-    {ok, So1, C1_00} = stream_test_utils:connect(Config, 1),
-    {ok, So2, C2_00} = stream_test_utils:connect(Config, 2),
+    {ok, So0, C0_00} = stream_test_utils:connect(Config, CL),
+    {ok, So1, C1_00} = stream_test_utils:connect(Config, CF1),
+    {ok, So2, C2_00} = stream_test_utils:connect(Config, CF2),
 
     C0_01 = register_sac(So0, C0_00, S, 0),
     C0_02 = receive_consumer_update(So0, C0_01),
@@ -196,27 +207,23 @@ simple_sac_consumer_should_get_disconnected_on_coord_leader_network_partition(Co
     assertSize(3, Consumers1),
     assertConsumersConnected(Consumers1),
 
-    N1 = nodename(Config, 0),
-    N2 = nodename(Config, 1),
-    N3 = nodename(Config, 2),
-
     %% N1 is the coordinator leader
-    Isolated = N1,
-    NotIsolated = N2,
+    Isolated = CL,
+    NotIsolated = CF1,
     {value, DisconnectedConsumer} =
         lists:search(fun(#consumer{pid = ConnPid}) ->
                              rpc(Config, erlang, node, [ConnPid]) =:= Isolated
                      end, Consumers1),
     #consumer{subscription_id = DiscSubId} = DisconnectedConsumer,
 
-    rabbit_ct_broker_helpers:block_traffic_between(Isolated, N2),
-    rabbit_ct_broker_helpers:block_traffic_between(Isolated, N3),
+    rabbit_ct_broker_helpers:block_traffic_between(Isolated, CF1),
+    rabbit_ct_broker_helpers:block_traffic_between(Isolated, CF2),
 
     wait_for_disconnected_consumer(Config, NotIsolated, S),
     wait_for_forgotten_consumer(Config, NotIsolated, S),
 
-    rabbit_ct_broker_helpers:allow_traffic_between(Isolated, N2),
-    rabbit_ct_broker_helpers:allow_traffic_between(Isolated, N3),
+    rabbit_ct_broker_helpers:allow_traffic_between(Isolated, CF1),
+    rabbit_ct_broker_helpers:allow_traffic_between(Isolated, CF2),
 
     wait_for_coordinator_ready(Config),
 
@@ -264,13 +271,107 @@ simple_sac_consumer_should_get_disconnected_on_coord_leader_network_partition(Co
 
     ok.
 
+super_stream_sac_consumer_should_get_disconnected_on_network_partition(Config) ->
+    init_coordinator(Config),
+    CL = coordinator_leader(Config),
+
+    Ss = rabbit_data_coercion:to_binary(?FUNCTION_NAME),
+    [Partition, _, _] = init_super_stream(Config, CL, Ss),
+    [L, F1, F2] = topology(Config, Partition),
+
+    wait_for_coordinator_ready(Config),
+
+    %% we expect the stream leader and the coordinator leader to be on the same node
+    %% another node will be isolated
+    ?assertEqual(L#node.name, coordinator_leader(Config)),
+
+    {ok, So0, C0_00} = stream_test_utils:connect(L#node.stream_port),
+    {ok, So1, C1_00} = stream_test_utils:connect(F1#node.stream_port),
+    {ok, So2, C2_00} = stream_test_utils:connect(F2#node.stream_port),
+
+    C0_01 = register_sac(So0, C0_00, Partition, 0, Ss),
+    C0_02 = receive_consumer_update(So0, C0_01),
+
+    C1_01 = register_sac(So1, C1_00, Partition, 1, Ss),
+    C2_01 = register_sac(So2, C2_00, Partition, 2, Ss),
+    SubIdToState0 = #{0 => {So0, C0_02},
+                      1 => {So1, C1_01},
+                      2 => {So2, C2_01}},
+
+    Consumers1 = query_consumers(Config, Partition),
+    assertSize(3, Consumers1),
+    assertConsumersConnected(Consumers1),
+
+    LN = L#node.name,
+    F1N = F1#node.name,
+    F2N = F2#node.name,
+
+    Isolated = F1N,
+    NotIsolated = F2N,
+    {value, DisconnectedConsumer} =
+        lists:search(fun(#consumer{pid = ConnPid}) ->
+                             rpc(Config, erlang, node, [ConnPid]) =:= Isolated
+                     end, Consumers1),
+    #consumer{subscription_id = DiscSubId} = DisconnectedConsumer,
+
+    rabbit_ct_broker_helpers:block_traffic_between(Isolated, LN),
+    rabbit_ct_broker_helpers:block_traffic_between(Isolated, F2N),
+
+    wait_for_disconnected_consumer(Config, NotIsolated, Partition),
+    wait_for_forgotten_consumer(Config, NotIsolated, Partition),
+
+    rabbit_ct_broker_helpers:allow_traffic_between(Isolated, LN),
+    rabbit_ct_broker_helpers:allow_traffic_between(Isolated, F2N),
+
+    wait_for_coordinator_ready(Config),
+
+    wait_for_all_consumers_connected(Config, NotIsolated, Partition),
+
+    Consumers2 = query_consumers(Config, NotIsolated, Partition),
+
+    %% the disconnected, then forgotten consumer is cancelled,
+    %% because the stream member on its node has been restarted
+    assertSize(2, Consumers2),
+    assertConsumersConnected(Consumers2),
+    assertEmpty(lists:filter(fun(C) ->
+                                     same_consumer(DisconnectedConsumer, C)
+                             end, Consumers2)),
+
+    SubIdToState1 =
+        maps:fold(fun(K, {S0, C0}, Acc) when K == DiscSubId ->
+                          %% cancelled consumer received a metadata update
+                          C1 = receive_metadata_update(S0, C0),
+                          Acc#{K => {S0, C1}};
+                     (K, {S0, C0}, Acc) ->
+                          Acc#{K => {S0, C0}}
+                  end, #{}, SubIdToState0),
+
+    delete_super_stream(L#node.stream_port, Ss),
+
+    %% online consumers should receive a metadata update frame (stream deleted)
+    %% we unqueue this frame before closing the connection
+    %% directly closing the connection of the cancelled consumer
+    maps:foreach(fun(K, {S0, C0}) when K /= DiscSubId ->
+                         {_, C1} = receive_commands(S0, C0),
+                         {ok, _} = stream_test_utils:close(S0, C1);
+                    (_, {S0, C0}) ->
+                         {ok, _} = stream_test_utils:close(S0, C0)
+                 end, SubIdToState1),
+    ok.
+
 super_stream_sac_consumer_should_get_disconnected_on_coord_leader_network_partition(Config) ->
+    init_coordinator(Config),
+    CL = coordinator_leader(Config),
     Ss = rabbit_data_coercion:to_binary(?FUNCTION_NAME),
-    [_, Partition, _] = init_super_stream(Config, Ss),
+    [_, Partition, _] = init_super_stream(Config, CL, Ss),
     [L, F1, F2] = topology(Config, Partition),
 
     wait_for_coordinator_ready(Config),
 
+    %% check stream leader and coordinator are not on the same node
+    %% the coordinator leader node will be isolated
+    ?assertNotEqual(L#node.name, coordinator_leader(Config)),
+
     {ok, So0, C0_00} = stream_test_utils:connect(L#node.stream_port),
     {ok, So1, C1_00} = stream_test_utils:connect(F1#node.stream_port),
     {ok, So2, C2_00} = stream_test_utils:connect(F2#node.stream_port),
@@ -406,13 +507,6 @@ leader(Members) ->
                       Acc
               end, undefined, Members).
 
-followers(Members) ->
-    maps:fold(fun(Node, {_, replica}, Acc) ->
-                      [Node | Acc];
-                 (_, _, Acc) ->
-                      Acc
-              end, [], Members).
-
 stream_members(Config, Stream) ->
     {ok, Q} = rpc(Config, rabbit_amqqueue, lookup, [Stream, <<"/">>]),
     #{name := StreamId} = amqqueue:get_type_state(Q),
@@ -421,8 +515,14 @@ stream_members(Config, Stream) ->
                         [StreamId, State]),
     Members.
 
-init_stream(Config, Node, St) ->
-    {ok, S, C0} = stream_test_utils:connect(Config, Node),
+init_coordinator(Config) ->
+    %% to make sure the coordinator is initialized
+    init_stream(Config, 0, <<"dummy">>),
+    delete_stream(stream_port(Config, 0), <<"dummy">>),
+    wait_for_coordinator_ready(Config).
+
+init_stream(Config, N, St) ->
+    {ok, S, C0} = stream_test_utils:connect(stream_port(Config, N)),
     {ok, C1} = stream_test_utils:create_stream(S, C0, St),
     NC = node_count(Config),
     wait_for_members(S, C1, St, NC),
@@ -433,8 +533,8 @@ delete_stream(Port, St) ->
     {ok, C1} = stream_test_utils:delete_stream(S, C0, St),
     {ok, _} = stream_test_utils:close(S, C1).
 
-init_super_stream(Config, Ss) ->
-    {ok, S, C0} = stream_test_utils:connect(Config, 0),
+init_super_stream(Config, Node, Ss) ->
+    {ok, S, C0} = stream_test_utils:connect(Config, Node),
     NC = node_count(Config),
     Partitions = [unicode:characters_to_binary([Ss, <<"-">>, integer_to_binary(N)])
                   || N <- lists:seq(0, NC - 1)],
@@ -509,9 +609,28 @@ query_consumers(Config, Node, Stream) ->
         rpc(Config, Node, rabbit_stream_coordinator, sac_state, []),
     Consumers.
 
+
+all_nodes(Config) ->
+    lists:map(fun(N) ->
+                      nodename(Config, N)
+              end, lists:seq(0, node_count(Config) - 1)).
+
 coordinator_status(Config) ->
     rpc(Config, rabbit_stream_coordinator, status, []).
 
+coordinator_leader(Config) ->
+    Status = coordinator_status(Config),
+    case lists:search(fun(St) ->
+                             RS = proplists:get_value(<<"Raft State">>, St,
+                                                      undefined),
+                             RS == leader
+                     end, Status) of
+        {value, Leader} ->
+            proplists:get_value(<<"Node Name">>, Leader, undefined);
+        _ ->
+            undefined
+    end.
+
 rpc(Config, M, F, A) ->
     rpc(Config, 0, M, F, A).