Merge branch 'master' into unavailable-qq-publish-fix

5 files changed, 315 insertions, 19 deletions

diff --git a/test/backing_queue_SUITE.erl b/test/backing_queue_SUITE.erl
index c3f87cce59..d262e4c513 100644
--- a/test/backing_queue_SUITE.erl
+++ b/test/backing_queue_SUITE.erl
@@ -733,7 +733,8 @@ bq_queue_recover1(Config) ->
     after 10000 -> exit(timeout_waiting_for_queue_death)
     end,
     rabbit_amqqueue:stop(?VHOST),
-    rabbit_amqqueue:start(rabbit_amqqueue:recover(?VHOST)),
+    {Recovered, [], []} = rabbit_amqqueue:recover(?VHOST),
+    rabbit_amqqueue:start(Recovered),
     {ok, Limiter} = rabbit_limiter:start_link(no_id),
     rabbit_amqqueue:with_or_die(
       QName,
diff --git a/test/channel_source_SUITE.erl b/test/channel_source_SUITE.erl
new file mode 100644
index 0000000000..56b287e913
--- /dev/null
+++ b/test/channel_source_SUITE.erl
@@ -0,0 +1,142 @@
+%% The contents of this file are subject to the Mozilla Public License
+%% Version 1.1 (the "License"); you may not use this file except in
+%% compliance with the License. You may obtain a copy of the License
+%% at http://www.mozilla.org/MPL/
+%%
+%% Software distributed under the License is distributed on an "AS IS"
+%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
+%% the License for the specific language governing rights and
+%% limitations under the License.
+%%
+%% The Original Code is RabbitMQ.
+%%
+%% The Initial Developer of the Original Code is GoPivotal, Inc.
+%% Copyright (c) 2007-2019 Pivotal Software, Inc.  All rights reserved.
+%%
+
+-module(channel_source_SUITE).
+
+-include_lib("amqp_client/include/amqp_client.hrl").
+
+-compile(export_all).
+
+all() ->
+    [
+      {group, non_parallel_tests}
+    ].
+
+groups() ->
+    [
+      {non_parallel_tests, [], [
+          network_rabbit_reader_channel_source,
+          network_arbitrary_channel_source,
+          direct_channel_source,
+          undefined_channel_source
+        ]}
+    ].
+
+%% -------------------------------------------------------------------
+%% Testsuite setup/teardown.
+%% -------------------------------------------------------------------
+
+init_per_suite(Config) ->
+    rabbit_ct_helpers:log_environment(),
+    rabbit_ct_helpers:run_setup_steps(Config).
+
+end_per_suite(Config) ->
+    rabbit_ct_helpers:run_teardown_steps(Config).
+
+init_per_group(_, Config) ->
+    Config.
+
+end_per_group(_, Config) ->
+    Config.
+
+init_per_testcase(Testcase, Config) ->
+    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()).
+
+end_per_testcase(Testcase, Config) ->
+    Config1 = rabbit_ct_helpers:run_steps(Config,
+      rabbit_ct_client_helpers:teardown_steps() ++
+      rabbit_ct_broker_helpers:teardown_steps()),
+    rabbit_ct_helpers:testcase_finished(Config1, Testcase).
+
+%% -------------------------------------------------------------------
+%% Testcases.
+%% -------------------------------------------------------------------
+
+network_rabbit_reader_channel_source(Config) ->
+    passed = rabbit_ct_broker_helpers:rpc(Config, 0,
+      ?MODULE, network_rabbit_reader_channel_source1, [Config]).
+
+network_rabbit_reader_channel_source1(Config) ->
+    ExistingChannels = rabbit_channel:list(),
+    Conn = rabbit_ct_client_helpers:open_unmanaged_connection(Config),
+    {ok, ClientCh} = amqp_connection:open_channel(Conn),
+    [ServerCh] = rabbit_channel:list() -- ExistingChannels,
+    [{source, rabbit_reader}] = rabbit_channel:info(ServerCh, [source]),
+    _ = rabbit_channel:source(ServerCh, ?MODULE),
+    [{source, ?MODULE}] = rabbit_channel:info(ServerCh, [source]),
+    amqp_channel:close(ClientCh),
+    amqp_connection:close(Conn),
+    {error, channel_terminated} = rabbit_channel:source(ServerCh, ?MODULE),
+    passed.
+
+network_arbitrary_channel_source(Config) ->
+    passed = rabbit_ct_broker_helpers:rpc(Config, 0,
+      ?MODULE, network_arbitrary_channel_source1, [Config]).
+
+network_arbitrary_channel_source1(Config) ->
+    Conn = rabbit_ct_client_helpers:open_unmanaged_connection(Config),
+    Writer = spawn(fun () -> rabbit_ct_broker_helpers:test_writer(self()) end),
+    {ok, Limiter} = rabbit_limiter:start_link(no_limiter_id),
+    {ok, Collector} = rabbit_queue_collector:start_link(no_collector_id),
+    {ok, Ch} = rabbit_channel:start_link(
+                 1, Conn, Writer, Conn, "", rabbit_framing_amqp_0_9_1,
+                 rabbit_ct_broker_helpers:user(<<"guest">>), <<"/">>, [],
+                 Collector, Limiter),
+    _ = rabbit_channel:source(Ch, ?MODULE),
+    [{amqp_params, #amqp_params_network{username = <<"guest">>,
+        password = <<"guest">>, host = "localhost", virtual_host = <<"/">>}}] =
+            rabbit_amqp_connection:amqp_params(Conn, 1000),
+    [{source, ?MODULE}] = rabbit_channel:info(Ch, [source]),
+    [exit(P, normal) || P <- [Writer, Limiter, Collector, Ch]],
+    amqp_connection:close(Conn),
+    {error, channel_terminated} = rabbit_channel:source(Ch, ?MODULE),
+    passed.
+
+direct_channel_source(Config) ->
+    passed = rabbit_ct_broker_helpers:rpc(Config, 0,
+      ?MODULE, direct_channel_source1, [Config]).
+
+direct_channel_source1(Config) ->
+    ExistingChannels = rabbit_channel:list(),
+    Conn = rabbit_ct_client_helpers:open_unmanaged_connection_direct(Config),
+    {ok, ClientCh} = amqp_connection:open_channel(Conn),
+    [ServerCh] = rabbit_channel:list() -- ExistingChannels,
+    [{source, rabbit_direct}] = rabbit_channel:info(ServerCh, [source]),
+    _ = rabbit_channel:source(ServerCh, ?MODULE),
+    [{source, ?MODULE}] = rabbit_channel:info(ServerCh, [source]),
+    amqp_channel:close(ClientCh),
+    amqp_connection:close(Conn),
+    {error, channel_terminated} = rabbit_channel:source(ServerCh, ?MODULE),
+    passed.
+
+undefined_channel_source(Config) ->
+    passed = rabbit_ct_broker_helpers:rpc(Config, 0,
+      ?MODULE, undefined_channel_source1, [Config]).
+
+undefined_channel_source1(_Config) ->
+    ExistingChannels = rabbit_channel:list(),
+    {_Writer, _Limiter, ServerCh} = rabbit_ct_broker_helpers:test_channel(),
+    [ServerCh] = rabbit_channel:list() -- ExistingChannels,
+    [{source, undefined}] = rabbit_channel:info(ServerCh, [source]),
+    _ = rabbit_channel:source(ServerCh, ?MODULE),
+    [{source, ?MODULE}] = rabbit_channel:info(ServerCh, [source]),
+    passed.
diff --git a/test/config_schema_SUITE_data/rabbit.snippets b/test/config_schema_SUITE_data/rabbit.snippets
index 0cd274b757..9afddd9b1e 100644
--- a/test/config_schema_SUITE_data/rabbit.snippets
+++ b/test/config_schema_SUITE_data/rabbit.snippets
@@ -580,19 +580,6 @@ credential_validator.regexp = ^abc\\d+",
      ]}],
    []},
 
-  {kernel_inet_dist_listen_min,
-   "inet_dist_listen_min = 16000",
-   [{kernel, [
-      {inet_dist_listen_min, 16000}
-     ]}],
-   []},
-  {kernel_inet_dist_listen_max,
-   "inet_dist_listen_max = 16100",
-   [{kernel, [
-      {inet_dist_listen_max, 16100}
-     ]}],
-   []},
-
   {log_syslog_settings,
    "log.syslog = true
     log.syslog.identity = rabbitmq
diff --git a/test/quorum_queue_SUITE.erl b/test/quorum_queue_SUITE.erl
index 164d37de4b..406c02de83 100644
--- a/test/quorum_queue_SUITE.erl
+++ b/test/quorum_queue_SUITE.erl
@@ -69,7 +69,8 @@ groups() ->
                                             delete_declare,
                                             metrics_cleanup_on_leadership_takeover,
                                             metrics_cleanup_on_leader_crash,
-                                            consume_in_minority
+                                            consume_in_minority,
+                                            shrink_all
                                             ]},
                       {cluster_size_5, [], [start_queue,
                                             start_queue_concurrent,
@@ -190,7 +191,8 @@ init_per_testcase(Testcase, Config) when Testcase == reconnect_consumer_and_publ
                                            [{rmq_nodes_count, 3},
                                             {rmq_nodename_suffix, Testcase},
                                             {tcp_ports_base},
-                                            {queue_name, Q}
+                                            {queue_name, Q},
+                                            {alt_queue_name, <<Q/binary, "_alt">>}
                                            ]),
     Config3 = rabbit_ct_helpers:run_steps(
                 Config2,
@@ -210,7 +212,8 @@ init_per_testcase(Testcase, Config) ->
     rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, delete_queues, []),
     Q = rabbit_data_coercion:to_binary(Testcase),
     Config2 = rabbit_ct_helpers:set_config(Config1,
-                                           [{queue_name, Q}
+                                           [{queue_name, Q},
+                                            {alt_queue_name, <<Q/binary, "_alt">>}
                                            ]),
     rabbit_ct_helpers:run_steps(Config2, rabbit_ct_client_helpers:setup_steps()).
 
@@ -622,7 +625,33 @@ consume_in_minority(Config) ->
 
     ?assertExit({{shutdown, {connection_closing, {server_initiated_close, 541, _}}}, _},
                 amqp_channel:call(Ch, #'basic.get'{queue = QQ,
-                                                   no_ack = false})).
+                                                   no_ack = false})),
+    ok = rabbit_ct_broker_helpers:start_node(Config, Server1),
+    ok = rabbit_ct_broker_helpers:start_node(Config, Server2),
+    ok.
+
+shrink_all(Config) ->
+    [Server0, Server1, Server2] =
+        rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+
+    Ch = rabbit_ct_client_helpers:open_channel(Config, Server0),
+    QQ = ?config(queue_name, Config),
+    AQ = ?config(alt_queue_name, Config),
+    ?assertEqual({'queue.declare_ok', QQ, 0, 0},
+                 declare(Ch, QQ, [{<<"x-queue-type">>, longstr, <<"quorum">>}])),
+    ?assertEqual({'queue.declare_ok', AQ, 0, 0},
+                 declare(Ch, AQ, [{<<"x-queue-type">>, longstr, <<"quorum">>}])),
+    timer:sleep(500),
+    Result = rpc:call(Server0, rabbit_quorum_queue, shrink_all, [Server2]),
+    ?assertMatch([{_, {ok, 2}}, {_, {ok, 2}}], Result),
+    Result1 = rpc:call(Server0, rabbit_quorum_queue, shrink_all, [Server1]),
+    ?assertMatch([{_, {ok, 1}}, {_, {ok, 1}}], Result1),
+    Result2 = rpc:call(Server0, rabbit_quorum_queue, shrink_all, [Server0]),
+    ?assertMatch([{_, {error, 1, last_node}},
+                  {_, {error, 1, last_node}}], Result2),
+    ok.
+
+
 
 subscribe_should_fail_when_global_qos_true(Config) ->
     [Server | _] = Servers = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
@@ -641,7 +670,7 @@ subscribe_should_fail_when_global_qos_true(Config) ->
         _ -> exit(subscribe_should_not_pass)
     catch
         _:_ = Err ->
-        ct:pal("Err ~p", [Err])
+        ct:pal("subscribe_should_fail_when_global_qos_true caught an error: ~p", [Err])
     end,
     ok.
 
diff --git a/test/rabbitmq_queues_cli_integration_SUITE.erl b/test/rabbitmq_queues_cli_integration_SUITE.erl
new file mode 100644
index 0000000000..1601ba82fe
--- /dev/null
+++ b/test/rabbitmq_queues_cli_integration_SUITE.erl
@@ -0,0 +1,137 @@
+%% The contents of this file are subject to the Mozilla Public License
+%% Version 1.1 (the "License"); you may not use this file except in
+%% compliance with the License. You may obtain a copy of the License
+%% at http://www.mozilla.org/MPL/
+%%
+%% Software distributed under the License is distributed on an "AS IS"
+%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
+%% the License for the specific language governing rights and
+%% limitations under the License.
+%%
+%% The Original Code is RabbitMQ.
+%%
+%% The Initial Developer of the Original Code is GoPivotal, Inc.
+%% Copyright (c) 2017-2019 Pivotal Software, Inc.  All rights reserved.
+%%
+-module(rabbitmq_queues_cli_integration_SUITE).
+
+-include_lib("common_test/include/ct.hrl").
+-include_lib("eunit/include/eunit.hrl").
+-include_lib("amqp_client/include/amqp_client.hrl").
+
+-compile(export_all).
+
+all() ->
+    [
+        {group, tests}
+    ].
+
+groups() ->
+    [
+        {tests, [], [
+            shrink,
+            grow,
+            grow_invalid_node_filtered
+        ]}
+    ].
+
+init_per_suite(Config) ->
+    rabbit_ct_helpers:log_environment(),
+    rabbit_ct_helpers:run_setup_steps(Config).
+
+end_per_suite(Config) ->
+    rabbit_ct_helpers:run_teardown_steps(Config).
+
+init_per_group(tests, Config0) ->
+    NumNodes = 3,
+    Config1 = rabbit_ct_helpers:set_config(
+                Config0, [{rmq_nodes_count, NumNodes},
+                          {rmq_nodes_clustered, true}]),
+    rabbit_ct_helpers:run_steps(Config1,
+                                rabbit_ct_broker_helpers:setup_steps() ++
+                                rabbit_ct_client_helpers:setup_steps()
+    ).
+
+end_per_group(tests, Config) ->
+    rabbit_ct_helpers:run_steps(Config,
+                                rabbit_ct_client_helpers:teardown_steps() ++
+                                    rabbit_ct_broker_helpers:teardown_steps()).
+
+init_per_testcase(Testcase, Config0) ->
+    rabbit_ct_helpers:ensure_rabbitmq_queues_cmd(
+      rabbit_ct_helpers:testcase_started(Config0, Testcase)).
+
+end_per_testcase(Testcase, Config0) ->
+    rabbit_ct_helpers:testcase_finished(Config0, Testcase).
+
+shrink(Config) ->
+    NodeConfig = rabbit_ct_broker_helpers:get_node_config(Config, 2),
+    Nodename2 = ?config(nodename, NodeConfig),
+    Ch = rabbit_ct_client_helpers:open_channel(Config, Nodename2),
+    %% declare a quorum queue
+    QName = "shrink1",
+    #'queue.declare_ok'{} = declare_qq(Ch, QName),
+    {ok, Out1} = rabbitmq_queues(Config, 0, ["shrink", Nodename2]),
+    ?assertMatch(#{{"/", "shrink1"} := {2, ok}}, parse_result(Out1)),
+    Nodename1 = rabbit_ct_broker_helpers:get_node_config(Config, 1, nodename),
+    {ok, Out2} = rabbitmq_queues(Config, 0, ["shrink", Nodename1]),
+    ?assertMatch(#{{"/", "shrink1"} := {1, ok}}, parse_result(Out2)),
+    Nodename0 = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
+    {ok, Out3} = rabbitmq_queues(Config, 0, ["shrink", Nodename0]),
+    ?assertMatch(#{{"/", "shrink1"} := {1, error}}, parse_result(Out3)),
+    ok.
+
+grow(Config) ->
+    NodeConfig = rabbit_ct_broker_helpers:get_node_config(Config, 2),
+    Nodename2 = ?config(nodename, NodeConfig),
+    Ch = rabbit_ct_client_helpers:open_channel(Config, Nodename2),
+    %% declare a quorum queue
+    QName = "grow1",
+    Args = [{<<"x-quorum-initial-group-size">>, long, 1}],
+    #'queue.declare_ok'{} = declare_qq(Ch, QName, Args),
+    Nodename0 = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
+    {ok, Out1} = rabbitmq_queues(Config, 0, ["grow", Nodename0, "all"]),
+    ?assertMatch(#{{"/", "grow1"} := {2, ok}}, parse_result(Out1)),
+    Nodename1 = rabbit_ct_broker_helpers:get_node_config(Config, 1, nodename),
+    {ok, Out2} = rabbitmq_queues(Config, 0, ["grow", Nodename1, "all"]),
+    ?assertMatch(#{{"/", "grow1"} := {3, ok}}, parse_result(Out2)),
+
+    {ok, Out3} = rabbitmq_queues(Config, 0, ["grow", Nodename0, "all"]),
+    ?assertNotMatch(#{{"/", "grow1"} := _}, parse_result(Out3)),
+    ok.
+
+grow_invalid_node_filtered(Config) ->
+    NodeConfig = rabbit_ct_broker_helpers:get_node_config(Config, 2),
+    Nodename2 = ?config(nodename, NodeConfig),
+    Ch = rabbit_ct_client_helpers:open_channel(Config, Nodename2),
+    %% declare a quorum queue
+    QName = "grow-err",
+    Args = [{<<"x-quorum-initial-group-size">>, long, 1}],
+    #'queue.declare_ok'{} = declare_qq(Ch, QName, Args),
+    DummyNode = not_really_a_node@nothing,
+    {ok, Out1} = rabbitmq_queues(Config, 0, ["grow", DummyNode, "all"]),
+    ?assertNotMatch(#{{"/", "grow-err"} := _}, parse_result(Out1)),
+    ok.
+
+parse_result(S) ->
+    Lines = string:split(S, "\n", all),
+    maps:from_list(
+      [{{Vhost, QName},
+        {erlang:list_to_integer(Size), case Result of
+                                           "ok" -> ok;
+                                           _ -> error
+                                       end}}
+       || [Vhost, QName, Size, Result] <-
+          [string:split(L, "\t", all) || L <- Lines]]).
+
+declare_qq(Ch, Q, Args0) ->
+    Args = [{<<"x-queue-type">>, longstr, <<"quorum">>}] ++ Args0,
+    amqp_channel:call(Ch, #'queue.declare'{queue = list_to_binary(Q),
+                                           durable = true,
+                                           auto_delete = false,
+                                           arguments = Args}).
+declare_qq(Ch, Q) ->
+    declare_qq(Ch, Q, []).
+
+rabbitmq_queues(Config, N, Args) ->
+    rabbit_ct_broker_helpers:rabbitmq_queues(Config, N, ["--silent" | Args]).