Merge pull request #14372 from rabbitmq/mergify/bp/v4.1.x/pr-14206

Collection of test fixes (2025Q2, batch 1) (backport #14206)

Author: michaelklishin

.github/workflows/test-make-target.yaml

@@ -119,5 +119,6 @@ jobs:
         name: CT logs (${{ inputs.plugin }} ${{ inputs.make_target }} OTP-${{ inputs.erlang_version }} ${{ inputs.metadata_store }}${{ inputs.mixed_clusters && ' mixed' || '' }})
         path: |
           logs/
+          deps/rabbitmq_cli/logs/
 #          !logs/**/log_private
         if-no-files-found: ignore

deps/rabbit/src/rabbit_vhosts.erl

@@ -124,7 +124,7 @@ start_processes_for_all(Nodes) ->
 
 -spec start_processes_for_all() -> 'ok'.
 start_processes_for_all() ->
-    start_processes_for_all(rabbit_nodes:list_reachable()).
+    start_processes_for_all(rabbit_nodes:list_running()).
 
 %% Same as rabbit_vhost_sup_sup:start_on_all_nodes/0.
 -spec start_on_all_nodes(vhost:name(), [node()]) -> 'ok'.

deps/rabbit/test/amqp_client_SUITE.erl

@@ -1773,6 +1773,11 @@ link_target_queue_deleted(QType, Config) ->
     ok = amqp10_client:send_msg(Sender, amqp10_msg:new(DTag1, <<"m1">>, false)),
     ok = wait_for_accepted(DTag1),
 
+    %% Load test module on the broker: we reference an anonymous function
+    %% from it during the configuration of meck.
+    [_ | _] = rabbit_ct_broker_helpers:rpc(
+                Config, ?MODULE, module_info, []),
+
     %% Mock delivery to the target queue to do nothing.
     rabbit_ct_broker_helpers:setup_meck(Config, [?MODULE]),
     Mod = rabbit_queue_type,
@@ -1833,6 +1838,11 @@ target_queues_deleted_accepted(Config) ->
     ok = amqp10_client:send_msg(Sender, amqp10_msg:new(DTag1, <<"m1">>, false)),
     ok = wait_for_accepted(DTag1),
 
+    %% Load test module on the broker: we reference an anonymous function
+    %% from it during the configuration of meck.
+    [_ | _] = rabbit_ct_broker_helpers:rpc(
+                Config, ?MODULE, module_info, []),
+
     %% Mock to deliver only to q1.
     rabbit_ct_broker_helpers:setup_meck(Config, [?MODULE]),
     Mod = rabbit_queue_type,
@@ -3979,7 +3989,7 @@ list_connections(Config) ->
     end,
 
     {ok, StdOut0} = rabbit_ct_broker_helpers:rabbitmqctl(Config, 0, ["list_connections", "--silent", "protocol"]),
-    Protocols0 = re:split(StdOut0, <<"\n">>, [trim]),
+    Protocols0 = re:split(string:trim(StdOut0), <<"\n">>, [trim]),
     %% Remove any whitespaces.
     Protocols1 = [binary:replace(Subject, <<" ">>, <<>>, [global]) || Subject <- Protocols0],
     Protocols = lists:sort(Protocols1),
@@ -3993,7 +4003,7 @@ list_connections(Config) ->
 
     %% CLI should list AMQP 1.0 container-id
     {ok, StdOut1} = rabbit_ct_broker_helpers:rabbitmqctl(Config, 0, ["list_connections", "--silent", "container_id"]),
-    ContainerIds0 = re:split(StdOut1, <<"\n">>, [trim]),
+    ContainerIds0 = re:split(string:trim(StdOut1), <<"\n">>, [trim]),
     ContainerIds = lists:sort(ContainerIds0),
     ?assertEqual([<<>>, ContainerId0, ContainerId2],
                  ContainerIds),
@@ -4630,6 +4640,11 @@ idle_time_out_on_server(Config) ->
         after 30000 -> ct:fail({missing_event, ?LINE})
         end,
 
+        %% Load test module on the broker: we reference an anonymous function
+        %% from it during the configuration of meck.
+        [_ | _] = rabbit_ct_broker_helpers:rpc(
+                    Config, ?MODULE, module_info, []),
+
         %% Mock the server socket to not have received any bytes.
         rabbit_ct_broker_helpers:setup_meck(Config),
         ok = rpc(Config, meck, new, [Mod, [no_link, passthrough]]),
@@ -4653,7 +4668,7 @@ idle_time_out_on_server(Config) ->
                   ct:fail({missing_event, ?LINE})
         end
     after
-        ?assert(rpc(Config, meck, validate, [Mod])),
+        _ = rpc(Config, meck, validate, [Mod]),
         ok = rpc(Config, meck, unload, [Mod]),
         ok = rpc(Config, application, set_env, [App, Par, DefaultVal])
     end.

deps/rabbit/test/amqp_jms_SUITE.erl

@@ -52,7 +52,7 @@ groups() ->
 
 suite() ->
     [
-     {timetrap, {minutes, 2}}
+     {timetrap, {minutes, 5}}
     ].
 
 init_per_suite(Config) ->

deps/rabbit/test/backing_queue_SUITE.erl

@@ -1445,18 +1445,18 @@ variable_queue_restart_large_seq_id2(VQ0, QName) ->
     Terms = variable_queue_read_terms(QName),
     Count = proplists:get_value(next_seq_id, Terms),
 
-    %% set a very high next_seq_id as if 100M messages have been
+    %% set a very high next_seq_id as if 100 billion messages have been
     %% published and consumed
-    Terms2 = lists:keyreplace(next_seq_id, 1, Terms, {next_seq_id, 100_000_000}),
+    Terms2 = lists:keyreplace(next_seq_id, 1, Terms, {next_seq_id, 100_000_000_000}),
 
     {TInit, VQ3} =
         timer:tc(
           fun() -> variable_queue_init(test_amqqueue(QName, true), Terms2) end,
           millisecond),
     %% even with a very high next_seq_id start of an empty queue
-    %% should be quick (few milliseconds, but let's give it 100ms, to
+    %% should be quick (few milliseconds, but let's give it 500ms, to
     %% avoid flaking on slow servers)
-    {true, _} = {TInit < 100, TInit},
+    {true, _} = {TInit < 500, TInit},
 
     %% should be empty now
     true = rabbit_variable_queue:is_empty(VQ3),

deps/rabbit/test/cluster_minority_SUITE.erl

@@ -9,14 +9,14 @@
 
 -include_lib("amqp_client/include/amqp_client.hrl").
 -include_lib("eunit/include/eunit.hrl").
+-include_lib("rabbitmq_ct_helpers/include/rabbit_assert.hrl").
 
 -compile([export_all, nowarn_export_all]).
 
 all() ->
     [
      {group, client_operations},
-     {group, cluster_operation_add},
-     {group, cluster_operation_remove}
+     {group, cluster_operation}
     ].
 
 groups() ->
@@ -42,8 +42,10 @@ groups() ->
                               delete_policy,
                               export_definitions
                              ]},
-     {cluster_operation_add, [], [add_node]},
-     {cluster_operation_remove, [], [remove_node]},
+     {cluster_operation, [], [add_node_when_seed_node_is_leader,
+                              add_node_when_seed_node_is_follower,
+                              remove_node_when_seed_node_is_leader,
+                              remove_node_when_seed_node_is_follower]},
      {feature_flags, [], [enable_feature_flag]}
     ].
 
@@ -127,26 +129,49 @@ init_per_group(Group, Config0) when Group == client_operations;
             partition_5_node_cluster(Config1),
             Config1
     end;
-init_per_group(Group, Config0) ->
+init_per_group(_Group, Config0) ->
     Config = rabbit_ct_helpers:set_config(Config0, [{rmq_nodes_count, 5},
-                                                    {rmq_nodename_suffix, Group},
                                                     {rmq_nodes_clustered, false},
                                                     {tcp_ports_base},
                                                     {net_ticktime, 5}]),
     Config1 = rabbit_ct_helpers:merge_app_env(
-                Config, {rabbit, [{forced_feature_flags_on_init, []}]}),
-    rabbit_ct_helpers:run_steps(Config1,
-                                rabbit_ct_broker_helpers:setup_steps() ++
-                                    rabbit_ct_client_helpers:setup_steps()).
+                Config, {rabbit, [{forced_feature_flags_on_init, []},
+                                  {khepri_leader_wait_retry_timeout, 30000}]}),
+    Config1.
 
-end_per_group(_, Config) ->
+end_per_group(Group, Config) when Group == client_operations;
+                                  Group == feature_flags ->
     rabbit_ct_helpers:run_steps(Config,
                                 rabbit_ct_client_helpers:teardown_steps() ++
-                                    rabbit_ct_broker_helpers:teardown_steps()).
-
+                                rabbit_ct_broker_helpers:teardown_steps());
+end_per_group(_Group, Config) ->
+    Config.
+
+init_per_testcase(Testcase, Config)
+  when Testcase =:= add_node_when_seed_node_is_leader orelse
+       Testcase =:= add_node_when_seed_node_is_follower orelse
+       Testcase =:= remove_node_when_seed_node_is_leader orelse
+       Testcase =:= remove_node_when_seed_node_is_follower ->
+    rabbit_ct_helpers:testcase_started(Config, Testcase),
+    Config1 = rabbit_ct_helpers:set_config(
+                Config, [{rmq_nodename_suffix, Testcase}]),
+    rabbit_ct_helpers:run_steps(
+      Config1,
+      rabbit_ct_broker_helpers:setup_steps() ++
+      rabbit_ct_client_helpers:setup_steps());
 init_per_testcase(Testcase, Config) ->
     rabbit_ct_helpers:testcase_started(Config, Testcase).
 
+end_per_testcase(Testcase, Config)
+  when Testcase =:= add_node_when_seed_node_is_leader orelse
+       Testcase =:= add_node_when_seed_node_is_follower orelse
+       Testcase =:= remove_node_when_seed_node_is_leader orelse
+       Testcase =:= remove_node_when_seed_node_is_follower ->
+    rabbit_ct_helpers:run_steps(
+      Config,
+      rabbit_ct_client_helpers:teardown_steps() ++
+      rabbit_ct_broker_helpers:teardown_steps()),
+    rabbit_ct_helpers:testcase_finished(Config, Testcase);
 end_per_testcase(Testcase, Config) ->
     rabbit_ct_helpers:testcase_finished(Config, Testcase).
 
@@ -271,53 +296,151 @@ set_policy(Config) ->
 delete_policy(Config) ->
     ?assertError(_, rabbit_ct_broker_helpers:clear_policy(Config, 0, <<"policy-to-delete">>)).
 
-add_node(Config) ->
-    [A, B, C, D, _E] = rabbit_ct_broker_helpers:get_node_configs(
+add_node_when_seed_node_is_leader(Config) ->
+    [A, B, C, _D, E] = rabbit_ct_broker_helpers:get_node_configs(
                          Config, nodename),
 
     %% Three node cluster: A, B, C
-    ok = rabbit_control_helper:command(stop_app, B),
-    ok = rabbit_control_helper:command(join_cluster, B, [atom_to_list(A)], []),
-    rabbit_control_helper:command(start_app, B),
+    Cluster = [A, B, C],
+    Config1 = rabbit_ct_broker_helpers:cluster_nodes(Config, Cluster),
 
-    ok = rabbit_control_helper:command(stop_app, C),
-    ok = rabbit_control_helper:command(join_cluster, C, [atom_to_list(A)], []),
-    rabbit_control_helper:command(start_app, C),
+    AMember = {rabbit_khepri:get_store_id(), A},
+    _ = ra:transfer_leadership(AMember, AMember),
+    clustering_utils:assert_cluster_status({Cluster, Cluster}, Cluster),
 
     %% Minority partition: A
+    partition_3_node_cluster(Config1),
+
+    Pong = ra:ping(AMember, 10000),
+    ct:pal("Member A state: ~0p", [Pong]),
+    case Pong of
+        {pong, State} when State =/= follower andalso State =/= candidate ->
+            Ret = rabbit_control_helper:command(
+                    join_cluster, E, [atom_to_list(A)], []),
+            ?assertMatch({error, _, _}, Ret),
+            {error, _, Msg} = Ret,
+            ?assertEqual(
+               match,
+               re:run(
+                 Msg, "(Khepri cluster could be in minority|\\{:rabbit, \\{\\{:error, :timeout\\})",
+                 [{capture, none}]));
+        Ret ->
+            ct:pal("A is not the expected leader: ~p", [Ret]),
+            {skip, "Node A was not elected leader"}
+    end.
+
+add_node_when_seed_node_is_follower(Config) ->
+    [A, B, C, _D, E] = rabbit_ct_broker_helpers:get_node_configs(
+                         Config, nodename),
+
+    %% Three node cluster: A, B, C
     Cluster = [A, B, C],
-    partition_3_node_cluster(Config),
-
-    ok = rabbit_control_helper:command(stop_app, D),
-    %% The command is appended to the log, but it will be dropped once the connectivity
-    %% is restored
-    ?assertMatch(ok,
-                 rabbit_control_helper:command(join_cluster, D, [atom_to_list(A)], [])),
-    timer:sleep(10000),
-    join_3_node_cluster(Config),
-    clustering_utils:assert_cluster_status({Cluster, Cluster}, Cluster).
-
-remove_node(Config) ->
+    Config1 = rabbit_ct_broker_helpers:cluster_nodes(Config, Cluster),
+
+    CMember = {rabbit_khepri:get_store_id(), C},
+    ra:transfer_leadership(CMember, CMember),
+    clustering_utils:assert_cluster_status({Cluster, Cluster}, Cluster),
+
+    %% Minority partition: A
+    partition_3_node_cluster(Config1),
+
+    AMember = {rabbit_khepri:get_store_id(), A},
+    Pong = ra:ping(AMember, 10000),
+    ct:pal("Member A state: ~0p", [Pong]),
+    case Pong of
+        {pong, State}
+          when State =:= follower orelse State =:= pre_vote ->
+            Ret = rabbit_control_helper:command(
+                    join_cluster, E, [atom_to_list(A)], []),
+            ?assertMatch({error, _, _}, Ret),
+            {error, _, Msg} = Ret,
+            ?assertEqual(
+               match,
+               re:run(
+                 Msg, "Khepri cluster could be in minority",
+                 [{capture, none}]));
+        {pong, await_condition} ->
+            Ret = rabbit_control_helper:command(
+                    join_cluster, E, [atom_to_list(A)], []),
+            ?assertMatch({error, _, _}, Ret),
+            {error, _, Msg} = Ret,
+            ?assertEqual(
+               match,
+               re:run(
+                 Msg, "\\{:rabbit, \\{\\{:error, :timeout\\}",
+                 [{capture, none}])),
+            clustering_utils:assert_cluster_status(
+              {Cluster, Cluster}, Cluster);
+        Ret ->
+            ct:pal("A is not the expected follower: ~p", [Ret]),
+            {skip, "Node A was not a follower"}
+    end.
+
+remove_node_when_seed_node_is_leader(Config) ->
     [A, B, C | _] = rabbit_ct_broker_helpers:get_node_configs(
                       Config, nodename),
 
     %% Three node cluster: A, B, C
-    ok = rabbit_control_helper:command(stop_app, B),
-    ok = rabbit_control_helper:command(join_cluster, B, [atom_to_list(A)], []),
-    rabbit_control_helper:command(start_app, B),
+    Cluster = [A, B, C],
+    Config1 = rabbit_ct_broker_helpers:cluster_nodes(Config, Cluster),
 
-    ok = rabbit_control_helper:command(stop_app, C),
-    ok = rabbit_control_helper:command(join_cluster, C, [atom_to_list(A)], []),
-    rabbit_control_helper:command(start_app, C),
+    AMember = {rabbit_khepri:get_store_id(), A},
+    ra:transfer_leadership(AMember, AMember),
+    clustering_utils:assert_cluster_status({Cluster, Cluster}, Cluster),
+    ct:pal("Waiting for cluster change permitted on node A"),
+    ?awaitMatch(
+       {ok, #{cluster_change_permitted := true,
+              leader_id := AMember}, AMember},
+       rabbit_ct_broker_helpers:rpc(
+         Config1, A, ra, member_overview, [AMember]),
+       60000),
+    {ok, Overview, AMember} = rabbit_ct_broker_helpers:rpc(
+                                Config1, A, ra, member_overview, [AMember]),
+    ct:pal("Member A overview: ~p", [maps:remove(machine, Overview)]),
 
     %% Minority partition: A
-    partition_3_node_cluster(Config),
+    partition_3_node_cluster(Config1),
+
+    ?assertEqual({pong, leader}, ra:ping(AMember, 10000)),
+    ?awaitMatch(
+       ok,
+       rabbit_control_helper:command(
+         forget_cluster_node, A, [atom_to_list(B)], []),
+       60000).
+
+remove_node_when_seed_node_is_follower(Config) ->
+    [A, B, C | _] = rabbit_ct_broker_helpers:get_node_configs(
+                      Config, nodename),
+
+    %% Three node cluster: A, B, C
     Cluster = [A, B, C],
+    Config1 = rabbit_ct_broker_helpers:cluster_nodes(Config, Cluster),
 
-    ok = rabbit_control_helper:command(forget_cluster_node, A, [atom_to_list(B)], []),
-    timer:sleep(10000),
-    join_3_node_cluster(Config),
-    clustering_utils:assert_cluster_status({Cluster, Cluster}, Cluster).
+    AMember = {rabbit_khepri:get_store_id(), A},
+    CMember = {rabbit_khepri:get_store_id(), C},
+    ra:transfer_leadership(CMember, CMember),
+    clustering_utils:assert_cluster_status({Cluster, Cluster}, Cluster),
+    ?awaitMatch(
+       {ok, #{cluster_change_permitted := true,
+              leader_id := CMember}, AMember},
+       rabbit_ct_broker_helpers:rpc(
+         Config1, A, ra, member_overview, [AMember]),
+       60000),
+    {ok, Overview, AMember} = rabbit_ct_broker_helpers:rpc(
+                                Config1, A, ra, member_overview, [AMember]),
+    ct:pal("Member A overview: ~p", [maps:remove(machine, Overview)]),
+
+    %% Minority partition: A
+    partition_3_node_cluster(Config1),
+
+    Ret = rabbit_control_helper:command(
+            forget_cluster_node, A, [atom_to_list(B)], []),
+    ?assertMatch({error, _, _}, Ret),
+    {error, _, Msg} = Ret,
+    ?assertEqual(
+       match,
+       re:run(
+         Msg, "Khepri cluster could be in minority", [{capture, none}])).
 
 enable_feature_flag(Config) ->
     [A | _] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),

deps/rabbit/test/clustering_management_SUITE.erl

@@ -745,13 +745,13 @@ is_in_minority(Ret) ->
     ?assertMatch(match, re:run(Msg, ".*timed out.*minority.*", [{capture, none}])).
 
 reset_last_disc_node(Config) ->
-    Servers = [Rabbit, Hare | _] = cluster_members(Config),
+    [Rabbit, Hare | _] = cluster_members(Config),
 
     stop_app(Config, Hare),
     ?assertEqual(ok, change_cluster_node_type(Config, Hare, ram)),
     start_app(Config, Hare),
 
-    case rabbit_ct_broker_helpers:enable_feature_flag(Config, Servers, khepri_db) of
+    case rabbit_ct_broker_helpers:enable_feature_flag(Config, [Rabbit], khepri_db) of
         ok ->
             %% The reset works after the switch to Khepri because the RAM node was
             %% implicitly converted to a disc one as Khepri always writes data on disc.