Merge pull request #1903 from rabbitmq/single-active-noconnection

rabbit_fifo: change single active consumer on noconnection

Author: michaelklishin

src/rabbit_fifo.erl

@@ -88,6 +88,8 @@
 
 -type consumer() :: #consumer{}.
 
+-type consumer_strategy() :: competing | single_active.
+
 -record(enqueuer,
         {next_seqno = 1 :: msg_seqno(),
          % out of order enqueues - sorted list
@@ -104,7 +106,8 @@
          max_length :: maybe(non_neg_integer()),
          max_bytes :: maybe(non_neg_integer()),
          %% whether single active consumer is on or not for this queue
-         consumer_strategy = default :: default | single_active,
+         consumer_strategy = competing :: consumer_strategy(),
+         %% the maximum number of unsuccessful delivery attempts permitted
          delivery_limit :: maybe(non_neg_integer())
         }).
 
@@ -114,7 +117,7 @@
          messages = #{} :: #{msg_in_id() => indexed_msg()},
          % defines the lowest message in id available in the messages map
          % that isn't a return
-         low_msg_num :: msg_in_id() | undefined,
+         low_msg_num :: maybe(msg_in_id()),
          % defines the next message in id to be added to the messages map
          next_msg_num = 1 :: msg_in_id(),
          % list of returned msg_in_ids - when checking out it picks from

src/rabbit_fifo.hrl

@@ -120,6 +120,7 @@ declare(Q) when ?amqqueue_is_quorum(Q) ->
             RaMachine = ra_machine(NewQ),
             ServerIds = [{RaName, Node} || Node <- Nodes],
             ClusterName = RaName,
+            TickTimeout = application:get_env(rabbit, quorum_tick_interval, ?TICK_TIMEOUT),
             RaConfs = [begin
                            UId = ra:new_uid(ra_lib:to_binary(ClusterName)),
                            FName = rabbit_misc:rs(QName),
@@ -129,7 +130,7 @@ declare(Q) when ?amqqueue_is_quorum(Q) ->
                              friendly_name => FName,
                              initial_members => ServerIds,
                              log_init_args => #{uid => UId},
-                             tick_timeout => ?TICK_TIMEOUT,
+                             tick_timeout => TickTimeout,
                              machine => RaMachine}
                        end || ServerId <- ServerIds],
 
@@ -190,7 +191,8 @@ update_consumer(QName, ChPid, ConsumerTag, Exclusive, AckRequired, Prefetch, Act
                                                QName, Prefetch, Active, ActivityStatus, Args).
 
 cancel_consumer_handler(QName, {ConsumerTag, ChPid}) ->
-    local_or_remote_handler(ChPid, rabbit_quorum_queue, cancel_consumer, [QName, ChPid, ConsumerTag]).
+    local_or_remote_handler(ChPid, rabbit_quorum_queue, cancel_consumer,
+                            [QName, ChPid, ConsumerTag]).
 
 cancel_consumer(QName, ChPid, ConsumerTag) ->
     catch rabbit_core_metrics:consumer_deleted(ChPid, ConsumerTag, QName),

src/rabbit_quorum_queue.erl

@@ -126,8 +126,10 @@ memory_tests() ->
 %% Testsuite setup/teardown.
 %% -------------------------------------------------------------------
 
-init_per_suite(Config) ->
+init_per_suite(Config0) ->
     rabbit_ct_helpers:log_environment(),
+    Config = rabbit_ct_helpers:merge_app_env(
+               Config0, {rabbit, [{quorum_tick_interval, 1000}]}),
     rabbit_ct_helpers:run_setup_steps(
       Config,
       [fun rabbit_ct_broker_helpers:enable_dist_proxy_manager/1]).

test/quorum_queue_SUITE.erl

@@ -271,21 +271,6 @@ scenario16(_Config) ->
                         delivery_limit => 1}, Commands),
     ok.
 
-fake_pid(_Config) ->
-    Pid = fake_external_pid(<<"mynode@banana">>),
-    ?assertNotEqual(node(Pid), node()),
-    ?assert(is_pid(Pid)),
-    ok.
-
-fake_external_pid(Node) when is_binary(Node) ->
-    ThisNodeSize = size(term_to_binary(node())) + 1,
-    Pid = spawn(fun () -> ok end),
-    %% drop the local node data from a local pid
-    <<_:ThisNodeSize/binary, LocalPidData/binary>> = term_to_binary(Pid),
-    S = size(Node),
-    %% replace it with the incoming node binary
-    Final = <<131,103, 100, 0, S, Node/binary, LocalPidData/binary>>,
-    binary_to_term(Final).
 
 snapshots(_Config) ->
     run_proper(
@@ -352,7 +337,7 @@ log_gen(Size) ->
 
 pid_gen(Nodes) ->
     ?LET(Node, oneof(Nodes),
-         fake_external_pid(atom_to_binary(Node, utf8))).
+         test_util:fake_pid(atom_to_binary(Node, utf8))).
 
 down_gen(Pid) ->
     ?LET(E, {down, Pid, oneof([noconnection, noproc])}, E).

test/rabbit_fifo_SUITE.erl

@@ -45,12 +45,14 @@ groups() ->
         ]}
     ].
 
-init_per_suite(Config) ->
+init_per_suite(Config0) ->
     rabbit_ct_helpers:log_environment(),
-    Config1 = rabbit_ct_helpers:set_config(Config, [
+    Config1 = rabbit_ct_helpers:set_config(Config0, [
         {rmq_nodename_suffix, ?MODULE}
     ]),
-    rabbit_ct_helpers:run_setup_steps(Config1,
+    Config = rabbit_ct_helpers:merge_app_env(
+               Config1, {rabbit, [{quorum_tick_interval, 1000}]}),
+    rabbit_ct_helpers:run_setup_steps(Config,
         rabbit_ct_broker_helpers:setup_steps() ++
         rabbit_ct_client_helpers:setup_steps()).
 

test/rabbit_fifo_prop_SUITE.erl

@@ -0,0 +1,17 @@
+-module(test_util).
+
+-export([
+         fake_pid/1
+         ]).
+
+
+fake_pid(Node) ->
+    NodeBin = rabbit_data_coercion:to_binary(Node),
+    ThisNodeSize = size(term_to_binary(node())) + 1,
+    Pid = spawn(fun () -> ok end),
+    %% drop the local node data from a local pid
+    <<_:ThisNodeSize/binary, LocalPidData/binary>> = term_to_binary(Pid),
+    S = size(NodeBin),
+    %% replace it with the incoming node binary
+    Final = <<131,103, 100, S:16/unsigned, NodeBin/binary, LocalPidData/binary>>,
+    binary_to_term(Final).