QQs: try an alternative process name in case of a conflict

deps/rabbit/docs/rabbitmq.conf.example

@@ -1013,6 +1013,35 @@
 # log.exchange.level = info
 
 
+## File size-based log rotation
+
+## Note that `log.file.rotation.size` cannot be combined with `log.file.rotation.date`,
+## the two options are mutually exclusive.
+
+## rotate when the file reaches 10 MiB
+# log.file.rotation.size = 10485760
+
+## keep up to 5 archived log files in addition to the current one
+# log.file.rotation.count = 5
+
+## compress the archived logs
+# log.file.rotation.compress = true
+
+
+## Date-based log rotation
+
+## Note that `log.file.rotation.date` cannot be combined with `log.file.rotation.size`,
+## the two options are mutually exclusive.
+
+## rotate every night at midnight
+# log.file.rotation.date = $D0
+
+## keep up to 5 archived log files in addition to the current one
+# log.file.rotation.count = 5
+
+## compress the archived logs
+# log.file.rotation.compress = true
+
 
 ## ----------------------------------------------------------------------------
 ## RabbitMQ LDAP Plugin

deps/rabbit/src/rabbit.erl

@@ -29,14 +29,17 @@
          base_product_name/0,
          base_product_version/0,
          motd_file/0,
-         motd/0]).
+         motd/0,
+         pg_local_scope/1]).
 %% For CLI, testing and mgmt-agent.
 -export([set_log_level/1, log_locations/0, config_files/0]).
 -export([is_booted/1, is_booted/0, is_booting/1, is_booting/0]).
 
 %%---------------------------------------------------------------------------
 %% Boot steps.
--export([maybe_insert_default_data/0, boot_delegate/0, recover/0, pg_local/0]).
+-export([maybe_insert_default_data/0, boot_delegate/0, recover/0,
+         pg_local_amqp_session/0,
+         pg_local_amqp_connection/0]).
 
 %% for tests
 -export([validate_msg_store_io_batch_size_and_credit_disc_bound/2]).
@@ -263,9 +266,15 @@
         {mfa,         {rabbit_vhosts, boot, []}},
         {requires,    notify_cluster}]}).
 
--rabbit_boot_step({pg_local,
-                   [{description, "local-only pg scope"},
-                    {mfa,         {rabbit, pg_local, []}},
+-rabbit_boot_step({pg_local_amqp_session,
+                   [{description, "local-only pg scope for AMQP sessions"},
+                    {mfa,         {rabbit, pg_local_amqp_session, []}},
+                    {requires,    kernel_ready},
+                    {enables,     core_initialized}]}).
+
+-rabbit_boot_step({pg_local_amqp_connection,
+                   [{description, "local-only pg scope for AMQP connections"},
+                    {mfa,         {rabbit, pg_local_amqp_connection, []}},
                     {requires,    kernel_ready},
                     {enables,     core_initialized}]}).
 
@@ -1115,11 +1124,18 @@ boot_delegate() ->
 -spec recover() -> 'ok'.
 
 recover() ->
-    ok = rabbit_vhost:recover(),
-    ok.
+    ok = rabbit_vhost:recover().
+
+pg_local_amqp_session() ->
+    PgScope = pg_local_scope(amqp_session),
+    rabbit_sup:start_child(pg_amqp_session, pg, [PgScope]).
+
+pg_local_amqp_connection() ->
+    PgScope = pg_local_scope(amqp_connection),
+    rabbit_sup:start_child(pg_amqp_connection, pg, [PgScope]).
 
-pg_local() ->
-    rabbit_sup:start_child(pg, [node()]).
+pg_local_scope(Prefix) ->
+    list_to_atom(io_lib:format("~s_~s", [Prefix, node()])).
 
 -spec maybe_insert_default_data() -> 'ok'.
 

deps/rabbit/src/rabbit_amqp1_0.erl

@@ -6,8 +6,6 @@
 %%
 -module(rabbit_amqp1_0).
 
--define(PROCESS_GROUP_NAME, rabbit_amqp10_connections).
-
 -export([list_local/0,
          register_connection/1]).
 
@@ -36,8 +34,11 @@ emit_connection_info_local(Items, Ref, AggregatorPid) ->
 
 -spec list_local() -> [pid()].
 list_local() ->
-    pg:get_local_members(node(), ?PROCESS_GROUP_NAME).
+    pg:which_groups(pg_scope()).
 
 -spec register_connection(pid()) -> ok.
 register_connection(Pid) ->
-    ok = pg:join(node(), ?PROCESS_GROUP_NAME, Pid).
+    ok = pg:join(pg_scope(), Pid, Pid).
+
+pg_scope() ->
+    rabbit:pg_local_scope(amqp_connection).

deps/rabbit/src/rabbit_amqp_session.erl

@@ -69,7 +69,6 @@
                    ?V_1_0_SYMBOL_MODIFIED]).
 -define(DEFAULT_EXCHANGE_NAME, <<>>).
 -define(PROTOCOL, amqp10).
--define(PROCESS_GROUP_NAME, amqp_sessions).
 -define(MAX_PERMISSION_CACHE_SIZE, 12).
 -define(HIBERNATE_AFTER, 6_000).
 -define(CREDIT_REPLY_TIMEOUT, 30_000).
@@ -373,8 +372,8 @@ init({ReaderPid, WriterPid, ChannelNum, MaxFrameSize, User, Vhost, ConnName,
                     handle_max = HandleMax0}}) ->
     process_flag(trap_exit, true),
     process_flag(message_queue_data, off_heap),
-    ok = pg:join(node(), ?PROCESS_GROUP_NAME, self()),
 
+    ok = pg:join(pg_scope(), self(), self()),
     Alarms0 = rabbit_alarm:register(self(), {?MODULE, conserve_resources, []}),
     Alarms = sets:from_list(Alarms0, [{version, 2}]),
 
@@ -439,7 +438,7 @@ terminate(_Reason, #state{incoming_links = IncomingLinks,
 
 -spec list_local() -> [pid()].
 list_local() ->
-    pg:get_local_members(node(), ?PROCESS_GROUP_NAME).
+    pg:which_groups(pg_scope()).
 
 -spec conserve_resources(pid(),
                          rabbit_alarm:resource_alarm_source(),
@@ -3432,6 +3431,9 @@ is_valid_max(Val) ->
     Val > 0 andalso
     Val =< ?UINT_MAX.
 
+pg_scope() ->
+    rabbit:pg_local_scope(amqp_session).
+
 -spec cap_credit(rabbit_queue_type:credit()) ->
     rabbit_queue_type:credit().
 cap_credit(DesiredCredit) ->

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -230,27 +230,48 @@ declare(Q, _Node) when ?amqqueue_is_quorum(Q) ->
 
 start_cluster(Q) ->
     QName = amqqueue:get_name(Q),
-    Durable = amqqueue:is_durable(Q),
-    AutoDelete = amqqueue:is_auto_delete(Q),
-    Arguments = amqqueue:get_arguments(Q),
     Opts = amqqueue:get_options(Q),
+    Arguments = amqqueue:get_arguments(Q),
     ActingUser = maps:get(user, Opts, ?UNKNOWN_USER),
     QuorumSize = get_default_quorum_initial_group_size(Arguments),
+    {LeaderNode, FollowerNodes} =
+        rabbit_queue_location:select_leader_and_followers(Q, QuorumSize),
     RaName = case qname_to_internal_name(QName) of
                  {ok, A} ->
                      A;
                  {error, {too_long, N}} ->
                      rabbit_data_coercion:to_atom(ra:new_uid(N))
              end,
-    {LeaderNode, FollowerNodes} =
-        rabbit_queue_location:select_leader_and_followers(Q, QuorumSize),
+    rabbit_log:debug("Will start up to ~w replicas for quorum ~ts with leader on node '~ts'",
+                     [QuorumSize, rabbit_misc:rs(QName), LeaderNode]),
+    case internal_declare(Q, RaName, LeaderNode, FollowerNodes) of
+        {error, cluster_not_formed, Queue} ->
+            _ = rabbit_amqqueue:internal_delete(Queue, ActingUser),
+            AlternativeRaName = rabbit_data_coercion:to_atom(ra:new_uid(rabbit_queue_type_util:name_concat(QName))),
+            rabbit_log:debug("Could not form cluster ~ts, trying alternative name ~ts",
+                     [RaName, AlternativeRaName]),
+            case internal_declare(Q, AlternativeRaName, LeaderNode, FollowerNodes) of
+                {ok, NewQ} -> 
+                    {new, NewQ};
+                {error, cluster_not_formed, NewQ} -> declare_queue_error(cluster_not_formed, NewQ, LeaderNode, ActingUser)
+            end;
+        {ok, NewQ} -> 
+            {new, NewQ}
+    end.
+
+internal_declare(Q, RaName, LeaderNode, FollowerNodes) ->
+    QName = amqqueue:get_name(Q),
+    Durable = amqqueue:is_durable(Q),
+    AutoDelete = amqqueue:is_auto_delete(Q),
+    Arguments = amqqueue:get_arguments(Q),
+    Opts = amqqueue:get_options(Q),
+    ActingUser = maps:get(user, Opts, ?UNKNOWN_USER),
+
     LeaderId = {RaName, LeaderNode},
     NewQ0 = amqqueue:set_pid(Q, LeaderId),
     NewQ1 = amqqueue:set_type_state(NewQ0,
                                     #{nodes => [LeaderNode | FollowerNodes]}),
 
-    rabbit_log:debug("Will start up to ~w replicas for quorum ~ts with leader on node '~ts'",
-                     [QuorumSize, rabbit_misc:rs(QName), LeaderNode]),
     case rabbit_amqqueue:internal_declare(NewQ1, false) of
         {created, NewQ} ->
             RaConfs = [make_ra_conf(NewQ, ServerId)
@@ -281,7 +302,9 @@ start_cluster(Q) ->
                                          {arguments, Arguments},
                                          {user_who_performed_action,
                                           ActingUser}]),
-                    {new, NewQ};
+                    {ok, NewQ};
+                {error, cluster_not_formed} -> 
+                    {error, cluster_not_formed, NewQ};
                 {error, Error} ->
                     declare_queue_error(Error, NewQ, LeaderNode, ActingUser)
             catch
@@ -291,7 +314,7 @@ start_cluster(Q) ->
         {existing, _} = Ex ->
             Ex
     end.
-
+ 
 declare_queue_error(Error, Queue, Leader, ActingUser) ->
     _ = rabbit_amqqueue:internal_delete(Queue, ActingUser),
     {protocol_error, internal_error,

deps/rabbit/test/quorum_queue_SUITE.erl

@@ -123,6 +123,7 @@ all_tests() ->
      consume_invalid_arg_2,
      start_queue,
      long_name,
+     conflicting_name,
      stop_queue,
      restart_queue,
      restart_all_types,
@@ -564,6 +565,38 @@ long_name(Config) ->
                          [{<<"x-queue-type">>, longstr, <<"quorum">>}])),
     ok.
 
+conflicting_name(Config) ->
+    Node = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
+    User = ?config(rmq_username, Config),
+    VHost1 = <<"foo">>,
+    VHost2 = <<"foo_bar">>,
+    QName1 = <<"bar_baz">>,
+    QName2 = <<"baz">>,
+
+    ok = rabbit_ct_broker_helpers:add_vhost(Config, Node, VHost1, User),
+    ok = rabbit_ct_broker_helpers:set_full_permissions(Config, User, VHost1),
+
+    ok = rabbit_ct_broker_helpers:add_vhost(Config, Node, VHost2, User),
+     ok = rabbit_ct_broker_helpers:set_full_permissions(Config, User, VHost2),
+
+    Conn1 = rabbit_ct_client_helpers:open_unmanaged_connection(Config, Node,
+                                                              VHost1),
+    {ok, Ch1} = amqp_connection:open_channel(Conn1),
+
+    ?assertEqual({'queue.declare_ok', QName1, 0, 0},
+                 declare(Ch1, QName1,
+                         [{<<"x-queue-type">>, longstr, <<"quorum">>}])),
+
+    Conn2 = rabbit_ct_client_helpers:open_unmanaged_connection(Config, Node,
+                                                              VHost2),
+    {ok, Ch2} = amqp_connection:open_channel(Conn2),
+
+    ?assertEqual({'queue.declare_ok', QName2, 0, 0},
+                 declare(Ch2, QName2,
+                         [{<<"x-queue-type">>, longstr, <<"quorum">>}])),
+
+    ok.
+
 start_queue_concurrent(Config) ->
     Servers = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
     LQ = ?config(queue_name, Config),

deps/rabbitmq_mqtt/src/rabbit_mqtt_sup.erl

@@ -29,7 +29,7 @@ init([{Listeners, SslListeners0}]) ->
           end,
     %% Use separate process group scope per RabbitMQ node. This achieves a local-only
     %% process group which requires less memory with millions of connections.
-    PgScope = list_to_atom(io_lib:format("~s_~s", [?PG_SCOPE, node()])),
+    PgScope = rabbit:pg_local_scope(?PG_SCOPE),
     persistent_term:put(?PG_SCOPE, PgScope),
     {ok,
      {#{strategy => one_for_all,