Merge branch 'main' into rabbitmq-server-8323

michaelklishin · web-flow · commit dcc7718592b1 · 2023-05-26T22:07:30.000+04:00
diff --git a/MODULE.bazel b/MODULE.bazel
@@ -179,8 +179,8 @@ erlang_package.hex_package(
 erlang_package.hex_package(
     name = "csv",
     build_file = "@rabbitmq-server//bazel:BUILD.csv",
-    sha256 = "54508938ac67e27966b10ef49606e3ad5995d665d7fc2688efb3eab1307c9079",
-    version = "2.4.1",
+    sha256 = "cbbe5455c93df5f3f2943e995e28b7a8808361ba34cf3e44267d77a01eaf1609",
+    version = "3.0.5",
 )
 
 erlang_package.hex_package(
@@ -258,13 +258,6 @@ erlang_package.git_package(
     tag = "v1.5.1",
 )
 
-erlang_package.hex_package(
-    name = "parallel_stream",
-    build_file = "@rabbitmq-server//bazel:BUILD.parallel_stream",
-    sha256 = "639b2e8749e11b87b9eb42f2ad325d161c170b39b288ac8d04c4f31f8f0823eb",
-    version = "1.0.6",
-)
-
 erlang_package.hex_package(
     name = "prometheus",
     build_file = "@rabbitmq-server//bazel:BUILD.prometheus",
@@ -359,7 +352,6 @@ use_repo(
     "json",
     "observer_cli",
     "osiris",
-    "parallel_stream",
     "prometheus",
     "ra",
     "ranch",
diff --git a/bazel/BUILD.csv b/bazel/BUILD.csv
@@ -1,3 +1,5 @@
+load("@rules_erlang//:erlang_bytecode2.bzl", "erlc_opts")
+
 filegroup(
     name = "sources",
     srcs = [
@@ -8,3 +10,17 @@ filegroup(
     ]),
     visibility = ["//visibility:public"],
 )
+
+erlc_opts(
+    name = "erlc_opts",
+    values = select({
+        "@rules_erlang//:debug_build": [
+            "+debug_info",
+        ],
+        "//conditions:default": [
+            "+debug_info",
+            "+deterministic",
+        ],
+    }),
+    visibility = [":__subpackages__"],
+)
diff --git a/bazel/BUILD.parallel_stream b/bazel/BUILD.parallel_stream
diff --git a/deps/rabbit/src/rabbit_quorum_queue.erl b/deps/rabbit/src/rabbit_quorum_queue.erl
@@ -1744,17 +1744,17 @@ force_shrink_member_to_current_member(VHost, Name) ->
 force_all_queues_shrink_member_to_current_member() ->
     rabbit_log:warning("Disaster recovery procedure: shrinking all quorum queues to a single node cluster"),
     Node = node(),
-    [begin
-         QName = amqqueue:get_name(Q),
-         {RaName, _} = amqqueue:get_pid(Q),
-         rabbit_log:warning("Disaster recovery procedure: shrinking queue ~p", [QName]),
-         ok = ra_server_proc:force_shrink_members_to_current_member({RaName, Node}),
-         Fun = fun (QQ) ->
-                       TS0 = amqqueue:get_type_state(QQ),
-                       TS = TS0#{nodes => [Node]},
-                       amqqueue:set_type_state(QQ, TS)
-               end,
-         _ = rabbit_amqqueue:update(QName, Fun)
-     end || Q <- rabbit_amqqueue:list(), amqqueue:get_type(Q) == ?MODULE],
+    _ = [begin
+             QName = amqqueue:get_name(Q),
+             {RaName, _} = amqqueue:get_pid(Q),
+             rabbit_log:warning("Disaster recovery procedure: shrinking queue ~p", [QName]),
+             ok = ra_server_proc:force_shrink_members_to_current_member({RaName, Node}),
+             Fun = fun (QQ) ->
+                           TS0 = amqqueue:get_type_state(QQ),
+                           TS = TS0#{nodes => [Node]},
+                           amqqueue:set_type_state(QQ, TS)
+                   end,
+             _ = rabbit_amqqueue:update(QName, Fun)
+         end || Q <- rabbit_amqqueue:list(), amqqueue:get_type(Q) == ?MODULE],
     rabbit_log:warning("Disaster recovery procedure: shrinking finished"),
     ok.
diff --git a/deps/rabbit/src/rabbit_vhost.erl b/deps/rabbit/src/rabbit_vhost.erl
@@ -252,9 +252,9 @@ update_metadata(Name, Metadata0, ActingUser) ->
             Error
     end.
 
--spec update(vhost:name(), binary(), [atom()], rabbit_types:username()) -> rabbit_types:ok_or_error(any()).
-update(Name, Description, Tags, ActingUser) ->
-    Metadata = #{description => Description, tags => Tags},
+-spec update(vhost:name(), binary(), [atom()], rabbit_queue_type:queue_type() | 'undefined', rabbit_types:username()) -> rabbit_types:ok_or_error(any()).
+update(Name, Description, Tags, DefaultQueueType, ActingUser) ->
+    Metadata = #{description => Description, tags => Tags, default_queue_type => DefaultQueueType},
     update_metadata(Name, Metadata, ActingUser).
 
 -spec delete(vhost:name(), rabbit_types:username()) -> rabbit_types:ok_or_error(any()).
@@ -325,7 +325,7 @@ put_vhost(Name, Description, Tags0, DefaultQueueType, Trace, Username) ->
     rabbit_log:debug("Parsed tags ~tp to ~tp", [Tags, ParsedTags]),
     Result = case exists(Name) of
                  true  ->
-                     update(Name, Description, ParsedTags, Username);
+                     update(Name, Description, ParsedTags, DefaultQueueType, Username);
                  false ->
                      Metadata0 = #{description => Description,
                                    tags => ParsedTags},
diff --git a/deps/rabbitmq_cli/BUILD.bazel b/deps/rabbitmq_cli/BUILD.bazel
@@ -10,21 +10,11 @@ load(
     "mix_archive_build",
 )
 
-mix_archive_build(
-    name = "parallel_stream_ez",
-    srcs = ["@parallel_stream//:sources"],
-    out = "parallel_stream.ez",
-    archives = ["@hex//:archive"],
-)
-
 mix_archive_build(
     name = "csv_ez",
     srcs = ["@csv//:sources"],
     out = "csv.ez",
     archives = ["@hex//:archive"],
-    ez_deps = [
-        ":parallel_stream_ez",
-    ],
 )
 
 mix_archive_build(
@@ -49,8 +39,7 @@ rabbitmqctl(
     license_files = glob(["LICENSE*"]),
     source_deps = {
         "@csv//:sources": "csv",
-        "@json//:sources": "json",
-        "@parallel_stream//:sources": "parallel_stream",  # transitive dep of csv 2.x
+        "@json//:sources": "json"
     },
     visibility = ["//visibility:public"],
     deps = [
@@ -113,7 +102,6 @@ plt(
         "runtime_tools",
     ],
     ez_deps = [
-        ":parallel_stream_ez",
         ":csv_ez",
         ":json_ez",
     ],
@@ -159,7 +147,6 @@ rabbitmqctl_test(
         "@csv//:sources": "csv",
         "@dialyxir//:sources": "dialyxir",
         "@json//:sources": "json",
-        "@parallel_stream//:sources": "parallel_stream",  # transitive dep of csv 2.x
         "@temp//:sources": "temp",
         "@x509//:sources": "x509",
     },
diff --git a/deps/rabbitmq_cli/Makefile b/deps/rabbitmq_cli/Makefile
@@ -1,13 +1,12 @@
 PROJECT = rabbitmq_cli
 
 BUILD_DEPS = rabbit_common
-DEPS = csv json observer_cli parallel_stream stdout_formatter
+DEPS = csv json observer_cli stdout_formatter
 TEST_DEPS = amqp amqp_client dialyxir temp x509 rabbit
 
 dep_amqp = hex 2.1.1
 dep_csv = hex 2.4.1
 dep_dialyxir = hex 0.5.1
-dep_parallel_stream = hex 1.0.6
 dep_json = hex 1.4.1
 dep_temp = hex 0.4.7
 dep_x509 = hex 0.7.0
diff --git a/deps/rabbitmq_cli/mix.exs b/deps/rabbitmq_cli/mix.exs
@@ -146,10 +146,6 @@ defmodule RabbitMQCtl.MixfileBase do
         :csv,
         path: Path.join(deps_dir, "csv")
       },
-      {
-        :parallel_stream,
-        path: Path.join(deps_dir, "parallel_stream"), override: true
-      },
       {
         :stdout_formatter,
         path: Path.join(deps_dir, "stdout_formatter"),
diff --git a/deps/rabbitmq_federation/src/rabbit_federation_link_util.erl b/deps/rabbitmq_federation/src/rabbit_federation_link_util.erl
@@ -140,13 +140,9 @@ connection_error(remote_start, {{shutdown, {server_initiated_close, Code, Messag
 connection_error(remote_start, E, Upstream, UParams, XorQName, State) ->
     rabbit_federation_status:report(
       Upstream, UParams, XorQName, clean_reason(E)),
-    Reason = case E of
-        {error, Value} -> Value;
-        Other -> Other
-    end,
     log_warning(XorQName, "did not connect to ~ts. Reason: ~tp",
                 [rabbit_federation_upstream:params_to_string(UParams),
-                 Reason]),
+                 E]),
     {stop, {shutdown, restart}, State};
 
 connection_error(remote, E, Upstream, UParams, XorQName, State) ->
diff --git a/deps/rabbitmq_federation/src/rabbit_federation_queue_link.erl b/deps/rabbitmq_federation/src/rabbit_federation_queue_link.erl
@@ -104,11 +104,7 @@ handle_cast(pause, State = #state{run = false}) ->
 handle_cast(pause, State = #not_started{}) ->
     {noreply, State#not_started{run = false}};
 
-handle_cast(pause, State = #state{ch = Ch, upstream = Upstream = #upstream{
-        name = UpName, queue_name = QName
-    }}) ->
-    rabbit_log_federation:debug("Federation link of ~s (upstream: '~s'): asked to pause",
-                                [QName, UpName]),
+handle_cast(pause, State = #state{ch = Ch, upstream = Upstream}) ->
     cancel(Ch, Upstream),
     {noreply, State#state{run = false}};
 
@@ -309,22 +305,18 @@ visit_match(_ ,_) ->
 consumer_tag(#upstream{consumer_tag = ConsumerTag}) ->
     ConsumerTag.
 
-consume(Ch, Upstream = #upstream{name = UpName}, UQueue) ->
+consume(Ch, Upstream, UQueue) ->
     ConsumerTag = consumer_tag(Upstream),
     NoAck = Upstream#upstream.ack_mode =:= 'no-ack',
-    rabbit_log_federation:debug("Federation link of ~ts: will consume from the upstream '~ts'",
-                                [rabbit_misc:rs(amqqueue:get_name(UQueue)), UpName]),
     amqp_channel:cast(
       Ch, #'basic.consume'{queue        = name(UQueue),
                            no_ack       = NoAck,
                            nowait       = true,
                            consumer_tag = ConsumerTag,
                            arguments    = [{<<"x-priority">>, long, -1}]}).
 
-cancel(Ch, Upstream = #upstream{name = UpName, queue_name = QName}) ->
+cancel(Ch, Upstream) ->
     ConsumerTag = consumer_tag(Upstream),
-    rabbit_log_federation:debug("Federation queue '~ts' link: will cancel consumer '~ts' on upstream '~ts'",
-                                [QName, ConsumerTag, UpName]),
     amqp_channel:cast(Ch, #'basic.cancel'{nowait       = true,
                                           consumer_tag = ConsumerTag}).
 
diff --git a/deps/rabbitmq_federation/src/rabbit_federation_upstream.erl b/deps/rabbitmq_federation/src/rabbit_federation_upstream.erl
@@ -61,7 +61,7 @@ params_table(SafeURI, XorQ) ->
 
 params_to_string(#upstream_params{safe_uri = SafeURI,
                                   x_or_q   = XorQ}) ->
-    print("~ts on '~ts'", [rabbit_misc:rs(r(XorQ)), SafeURI]).
+    print("~ts on ~ts", [rabbit_misc:rs(r(XorQ)), SafeURI]).
 
 remove_credentials(URI) ->
     list_to_binary(amqp_uri:remove_credentials(binary_to_list(URI))).
diff --git a/deps/rabbitmq_web_mqtt/src/rabbit_web_mqtt_app.erl b/deps/rabbitmq_web_mqtt/src/rabbit_web_mqtt_app.erl
@@ -143,20 +143,20 @@ start_tls_listener(TLSConf0, CowboyOpts) ->
 
 listener_started(Protocol, Listener) ->
     Port = rabbit_misc:pget(port, Listener),
-    case rabbit_misc:pget(ip, Listener) of
-        undefined ->
-            [rabbit_networking:tcp_listener_started(Protocol, Listener,
-                                                    IPAddress, Port)
-             || {IPAddress, _Port, _Family}
-                    <- rabbit_networking:tcp_listener_addresses(Port)];
-        IP when is_tuple(IP) ->
-            rabbit_networking:tcp_listener_started(Protocol, Listener,
-                                                   IP, Port);
-        IP when is_list(IP) ->
-            {ok, ParsedIP} = inet_parse:address(IP),
-            rabbit_networking:tcp_listener_started(Protocol, Listener,
-                                                   ParsedIP, Port)
-    end,
+    _ = case rabbit_misc:pget(ip, Listener) of
+            undefined ->
+                [rabbit_networking:tcp_listener_started(Protocol, Listener,
+                                                        IPAddress, Port)
+                 || {IPAddress, _Port, _Family}
+                        <- rabbit_networking:tcp_listener_addresses(Port)];
+            IP when is_tuple(IP) ->
+                rabbit_networking:tcp_listener_started(Protocol, Listener,
+                                                       IP, Port);
+            IP when is_list(IP) ->
+                {ok, ParsedIP} = inet_parse:address(IP),
+                rabbit_networking:tcp_listener_started(Protocol, Listener,
+                                                       ParsedIP, Port)
+        end,
     ok.
 
 get_tcp_conf(TCPConf0) ->
diff --git a/deps/rabbitmq_web_stomp/src/rabbit_web_stomp_listener.erl b/deps/rabbitmq_web_stomp/src/rabbit_web_stomp_listener.erl
@@ -165,20 +165,20 @@ start_tls_listener(TLSConf0, CowboyOpts0, Routes) ->
 
 listener_started(Protocol, Listener) ->
     Port = rabbit_misc:pget(port, Listener),
-    case rabbit_misc:pget(ip, Listener) of
-        undefined ->
-            [rabbit_networking:tcp_listener_started(Protocol, Listener,
-                                                    IPAddress, Port)
-             || {IPAddress, _Port, _Family}
-                    <- rabbit_networking:tcp_listener_addresses(Port)];
-        IP when is_tuple(IP) ->
-            rabbit_networking:tcp_listener_started(Protocol, Listener,
-                                                   IP, Port);
-        IP when is_list(IP) ->
-            {ok, ParsedIP} = inet_parse:address(IP),
-            rabbit_networking:tcp_listener_started(Protocol, Listener,
-                                                   ParsedIP, Port)
-    end,
+    _ = case rabbit_misc:pget(ip, Listener) of
+            undefined ->
+                [rabbit_networking:tcp_listener_started(Protocol, Listener,
+                                                        IPAddress, Port)
+                 || {IPAddress, _Port, _Family}
+                        <- rabbit_networking:tcp_listener_addresses(Port)];
+            IP when is_tuple(IP) ->
+                rabbit_networking:tcp_listener_started(Protocol, Listener,
+                                                       IP, Port);
+            IP when is_list(IP) ->
+                {ok, ParsedIP} = inet_parse:address(IP),
+                rabbit_networking:tcp_listener_started(Protocol, Listener,
+                                                       ParsedIP, Port)
+        end,
     ok.
 
 get_env(Key, Default) ->
diff --git a/dist.bzl b/dist.bzl
@@ -351,23 +351,13 @@ def source_archive(
         prefix = "deps/csv",
     )
 
-    pkg_files(
-        name = "parallel_stream-files",
-        srcs = [
-            "@parallel_stream//:sources",
-        ],
-        strip_prefix = "",
-        prefix = "deps/parallel_stream",
-    )
-
     pkg_tar(
         name = name,
         extension = "tar.xz",
         srcs = [
             ":deps-files",
             ":json-files",
             ":csv-files",
-            ":parallel_stream-files",
             Label("@rabbitmq-server//:root-licenses"),
         ],
         visibility = ["//visibility:public"],
diff --git a/rabbitmq-components.mk b/rabbitmq-components.mk
@@ -114,7 +114,7 @@ dep_accept = hex 0.3.5
 dep_cowboy = hex 2.10.0
 dep_cowlib = hex 2.12.1
 dep_credentials_obfuscation = hex 3.4.0
-dep_looking_glass = git https://github.com/rabbitmq/looking_glass.git master
+dep_looking_glass = git https://github.com/rabbitmq/looking_glass.git main
 dep_prometheus = hex 4.10.0
 dep_ra = hex 2.6.1
 dep_ranch = hex 2.1.0
diff --git a/release-notes/3.12.0.md b/release-notes/3.12.0.md
@@ -243,6 +243,12 @@ This release includes all bug fixes shipped in the `3.11.x` series.
 
    GitHub issue: [#7743](https://github.com/rabbitmq/rabbitmq-server/issues/7743)
 
+ * Quorum queues now have a mechanism of shrinking down to just one replica.
+   This is meant to be used **exclusively** for disaster recovery when a majority
+   of nodes hosting replicas of a queue were permanently lost.
+
+   GitHub issue: [#8322](https://github.com/rabbitmq/rabbitmq-server/pull/8322)
+
  * Classic mirrored queues that had a [length limit](https://rabbitmq.com/maxlength.html) defined on them
    handled the overflow of messages differently from the current elected leader,
    eventually causing exceptions and replica restarts.
diff --git a/release-notes/3.13.0.md b/release-notes/3.13.0.md
