Merge pull request #9026 from rabbitmq/fixes-from-the-khepri-branch

Fixes from the `khepri` branch

Author: michaelklishin

deps/rabbit/src/rabbit.erl

@@ -1038,10 +1038,7 @@ prep_stop(State) ->
 
 stop(State) ->
     ok = rabbit_alarm:stop(),
-    ok = case rabbit_db_cluster:is_clustered() of
-             true  -> ok;
-             false -> rabbit_table:clear_ram_only_tables()
-         end,
+    ok = rabbit_table:maybe_clear_ram_only_tables(),
     case State of
         [] -> rabbit_prelaunch:set_stop_reason(normal);
         _  -> rabbit_prelaunch:set_stop_reason(State)

deps/rabbit/src/rabbit_db_cluster.erl

@@ -100,6 +100,7 @@ join(RemoteNode, NodeType)
     end.
 
 join_using_mnesia(ClusterNodes, NodeType) when is_list(ClusterNodes) ->
+    ok = rabbit_mnesia:reset_gracefully(),
     rabbit_mnesia:join_cluster(ClusterNodes, NodeType).
 
 -spec forget_member(Node, RemoveWhenOffline) -> ok when

deps/rabbit/src/rabbit_mnesia.erl

@@ -47,7 +47,9 @@
          on_node_down/1,
 
          %% Helpers for diagnostics commands
-         schema_info/1
+         schema_info/1,
+
+         reset_gracefully/0
         ]).
 
 %% Mnesia queries

deps/rabbit/src/rabbit_table.erl

@@ -11,8 +11,9 @@
     create/0, create/2, ensure_local_copies/1, ensure_table_copy/3,
     wait_for_replicated/1, wait/1, wait/2,
     force_load/0, is_present/0, is_empty/0, needs_default_data/0,
-    check_schema_integrity/1, clear_ram_only_tables/0, retry_timeout/0,
-    wait_for_replicated/0]).
+    check_schema_integrity/1,
+    clear_ram_only_tables/0, maybe_clear_ram_only_tables/0,
+    retry_timeout/0, wait_for_replicated/0]).
 
 %% for testing purposes
 -export([definitions/0]).
@@ -191,6 +192,12 @@ clear_ram_only_tables() ->
       end, names()),
     ok.
 
+maybe_clear_ram_only_tables() ->
+    ok = case rabbit_db_cluster:is_clustered() of
+             true  -> ok;
+             false -> clear_ram_only_tables()
+         end.
+
 %% The sequence in which we delete the schema and then the other
 %% tables is important: if we delete the schema first when moving to
 %% RAM mnesia will loudly complain since it doesn't make much sense to

deps/rabbit/test/peer_discovery_classic_config_SUITE.erl

@@ -94,12 +94,10 @@ init_per_testcase(successful_discovery_with_a_subset_of_nodes_coming_online = Te
     %% not exist and not just unreachable
     Config3 = rabbit_ct_helpers:merge_app_env(Config2,
       {rabbit, [
-          {cluster_formation, [
-              {discovery_retry_limit, 10},
-              {discovery_retry_interval, 200}
-          ]},
           {cluster_nodes, {NodeNamesWithHostname, disc}},
           {cluster_formation, [
+              {discovery_retry_limit, 2},
+              {discovery_retry_interval, 100},
               {internal_lock_retries, 10}
           ]}
       ]}),

deps/rabbit/test/per_vhost_connection_limit_SUITE.erl

@@ -657,6 +657,17 @@ cluster_multiple_vhosts_zero_limit(Config) ->
 vhost_limit_after_node_renamed(Config) ->
     A = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
 
+    %% Make sure the maintenance mode states Mnesia table is replicated
+    %% everywhere. We do this here, just in case mixed-version testing is
+    %% against a version of RabbitMQ that doesn't have the fix yet.
+    %%
+    %% See https://github.com/rabbitmq/rabbitmq-server/pull/9005.
+    B = rabbit_ct_broker_helpers:get_node_config(Config, 1, nodename),
+    rabbit_ct_broker_helpers:rpc(
+      Config, B,
+      rabbit_table, ensure_table_copy,
+      [rabbit_node_maintenance_states, B, ram_copies]),
+
     VHost = <<"/renaming_node">>,
     set_up_vhost(Config, VHost),
     set_vhost_connection_limit(Config, VHost, 2),

deps/rabbitmq_ct_helpers/src/rabbit_ct_broker_helpers.erl

@@ -1080,7 +1080,60 @@ rabbitmqctl(Config, Node, Args) ->
     rabbitmqctl(Config, Node, Args, infinity).
 
 rabbitmqctl(Config, Node, Args, Timeout) ->
-    Rabbitmqctl = ?config(rabbitmqctl_cmd, Config),
+    %% We want to use the CLI from the given node if there is a secondary
+    %% umbrella being configured.
+    I = get_node_index(Config, Node),
+    CanUseSecondary = (I + 1) rem 2 =:= 0,
+    BazelRunSecCmd = rabbit_ct_helpers:get_config(
+                       Config, rabbitmq_run_secondary_cmd),
+    UseSecondaryUmbrella = case ?config(secondary_umbrella, Config) of
+                               false ->
+                                   case BazelRunSecCmd of
+                                       undefined -> false;
+                                       _         -> CanUseSecondary
+                                   end;
+                               _ ->
+                                   CanUseSecondary
+                           end,
+    Rabbitmqctl = case UseSecondaryUmbrella of
+                      true ->
+                          case BazelRunSecCmd of
+                              undefined ->
+                                  SrcDir = ?config(
+                                              secondary_rabbit_srcdir,
+                                              Config),
+                                  SecDepsDir = ?config(
+                                                  secondary_erlang_mk_depsdir,
+                                                  Config),
+                                  SecNewScriptsDir = filename:join(
+                                                       [SecDepsDir,
+                                                        SrcDir,
+                                                        "sbin"]),
+                                  SecOldScriptsDir = filename:join(
+                                                       [SecDepsDir,
+                                                        "rabbit",
+                                                        "scripts"]),
+                                  SecNewScriptsDirExists = filelib:is_dir(
+                                                             SecNewScriptsDir),
+                                  SecScriptsDir =
+                                  case SecNewScriptsDirExists of
+                                      true  -> SecNewScriptsDir;
+                                      false -> SecOldScriptsDir
+                                  end,
+                                  rabbit_misc:format(
+                                    "~ts/rabbitmqctl", [SecScriptsDir]);
+                              _ ->
+                                  BazelSecScriptsDir = filename:dirname(
+                                                         BazelRunSecCmd),
+                                  filename:join(
+                                    [BazelSecScriptsDir,
+                                     "sbin",
+                                     "rabbitmqctl"])
+                          end;
+                      false ->
+                          ?config(rabbitmqctl_cmd, Config)
+                  end,
+
     NodeConfig = get_node_config(Config, Node),
     Nodename = ?config(nodename, NodeConfig),
     Env0 = [
@@ -1138,6 +1191,23 @@ rabbitmq_queues(Config, Node, Args) ->
 %% Other helpers.
 %% -------------------------------------------------------------------
 
+get_node_index(Config, Node) when is_atom(Node) andalso Node =/= undefined ->
+    NodeConfigs = get_node_configs(Config),
+    get_node_index1(NodeConfigs, Node, 0);
+get_node_index(_Config, I) when is_integer(I) andalso I >= 0 ->
+    I.
+
+get_node_index1([NodeConfig | Rest], Node, I) ->
+    case ?config(nodename, NodeConfig) of
+        Node ->
+            I;
+        _ ->
+            case ?config(initial_nodename, NodeConfig) of
+                Node -> I;
+                _    -> get_node_index1(Rest, Node, I + 1)
+            end
+    end.
+
 get_node_configs(Config) ->
     ?config(rmq_nodes, Config).
 

deps/rabbitmq_mqtt/test/reader_SUITE.erl

@@ -243,7 +243,7 @@ event_authentication_failure(Config) ->
 
     ?assertMatch({error, _}, emqtt:connect(C)),
 
-    [E] = util:get_events(Server),
+    [E | _] = util:get_events(Server, user_authentication_failure),
     util:assert_event_type(user_authentication_failure, E),
     util:assert_event_prop([{name, <<"Trudy">>},
                             {connection_type, network}],

deps/rabbitmq_mqtt/test/util.erl

@@ -19,6 +19,7 @@
          connect/4,
          start_client/4,
          get_events/1,
+         get_events/2,
          assert_event_type/2,
          assert_event_prop/2,
          assert_message_expiry_interval/2,
@@ -104,6 +105,11 @@ get_events(Node) ->
     ?assert(is_list(Result)),
     Result.
 
+get_events(Node, Type) ->
+    lists:filter(fun(#event{type = T}) ->
+                         T == Type
+                 end, get_events(Node)).
+
 assert_event_type(ExpectedType, #event{type = ActualType}) ->
     ?assertEqual(ExpectedType, ActualType).