Ack messages consumed from quorum queues

[#154472221]

Author: dcorbacho

src/rabbit_amqqueue.erl

@@ -992,8 +992,18 @@ purge(#amqqueue{ pid = QPid }) ->
 requeue(QPid, MsgIds, ChPid) ->
     delegate:invoke(QPid, {gen_server2, call, [{requeue, MsgIds, ChPid}, infinity]}).
 
-ack(QPid, MsgIds, ChPid) ->
-    delegate:invoke_no_result(QPid, {gen_server2, cast, [{ack, MsgIds, ChPid}]}).
+ack(QPid, MsgIds, ChPid) when is_pid(QPid) ->
+    delegate:invoke_no_result(QPid, {gen_server2, cast, [{ack, MsgIds, ChPid}]});
+ack(Id, CTagsMsgIds, ChPid) ->
+    %% TODO Basic get doesn't have a consumer tag, we have to fake it again
+    [{ok, _, _} = ra:send_and_await_consensus(Id, {settle, MsgId, {quorum_ctag(CTag), ChPid}})
+     || {CTag, MsgId} <- CTagsMsgIds],
+    ok.
+
+quorum_ctag(none) ->
+    <<"ctag">>;
+quorum_ctag(Other) ->
+    Other.
 
 reject(QPid, Requeue, MsgIds, ChPid) ->
     delegate:invoke_no_result(QPid, {gen_server2, cast, [{reject, Requeue, MsgIds, ChPid}]}).

src/rabbit_channel.erl

@@ -616,7 +616,7 @@ handle_info({ra_fifo, {RegisteredName, _} = Id, {delivery, CTag, MsgId, Msg}},
             ok
     end,
     noreply(handle_deliver(CTag, AckRequired,
-                           {Id, whereis(RegisteredName), MsgId, true, Msg}, State));
+                           {RegisteredName, Id, MsgId, true, Msg}, State));
 
 handle_info({bump_credit, Msg}, State) ->
     %% A rabbit_amqqueue_process is granting credit to our channel. If
@@ -1782,13 +1782,17 @@ notify_queues(State = #ch{consumer_mapping  = Consumers,
 
 foreach_per_queue(_F, []) ->
     ok;
-foreach_per_queue(F, [{_DTag, _CTag, {QPid, MsgId}}]) -> %% common case
+foreach_per_queue(F, [{_DTag, _CTag, {QPid, MsgId}}]) when is_pid(QPid) -> %% common case
     F(QPid, [MsgId]);
+foreach_per_queue(F, [{_DTag, CTag, {QPid, MsgId}}]) -> %% quorum queue, needs the consumer tag
+    F(QPid, [{CTag, MsgId}]);
 %% NB: UAL should be in youngest-first order; the tree values will
 %% then be in oldest-first order
 foreach_per_queue(F, UAL) ->
-    T = lists:foldl(fun ({_DTag, _CTag, {QPid, MsgId}}, T) ->
-                            rabbit_misc:gb_trees_cons(QPid, MsgId, T)
+    T = lists:foldl(fun ({_DTag, _CTag, {QPid, MsgId}}, T) when is_pid(QPid) ->
+                            rabbit_misc:gb_trees_cons(QPid, MsgId, T);
+                        ({_DTag, CTag, {QPid, MsgId}}, T) ->
+                            rabbit_misc:gb_trees_cons(QPid, {CTag, MsgId}, T)
                     end, gb_trees:empty(), UAL),
     rabbit_misc:gb_trees_foreach(F, T).
 

test/quorum_queue_SUITE.erl

@@ -42,7 +42,9 @@ groups() ->
           consume_from_empty_queue,
           consume_and_autoack_from_queue,
           subscribe_to_queue,
-          subscribe_with_autoack_to_queue
+          subscribe_with_autoack_to_queue,
+          consume_and_ack_from_queue,
+          subscribe_and_ack_to_queue
         ]}
     ].
 
@@ -270,8 +272,6 @@ publish_and_restart(Config) ->
     wait_for_messages(Config, QQ, <<"2">>, <<"2">>, <<"0">>).
 
 consume_from_queue(Config) ->
-    %% Test the node restart with both types of queues (quorum and classic) to
-    %% ensure there are no regressions
     Node = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
 
     Ch = rabbit_ct_client_helpers:open_channel(Config, Node),
@@ -287,8 +287,6 @@ consume_from_queue(Config) ->
     wait_for_messages(Config, QQ, <<"1">>, <<"1">>, <<"0">>).
 
 consume_and_autoack_from_queue(Config) ->
-    %% Test the node restart with both types of queues (quorum and classic) to
-    %% ensure there are no regressions
     Node = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
 
     Ch = rabbit_ct_client_helpers:open_channel(Config, Node),
@@ -304,8 +302,6 @@ consume_and_autoack_from_queue(Config) ->
     wait_for_messages(Config, QQ, <<"0">>, <<"0">>, <<"0">>).
 
 consume_from_empty_queue(Config) ->
-    %% Test the node restart with both types of queues (quorum and classic) to
-    %% ensure there are no regressions
     Node = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
 
     Ch = rabbit_ct_client_helpers:open_channel(Config, Node),
@@ -316,8 +312,6 @@ consume_from_empty_queue(Config) ->
     consume_empty(Ch, QQ, false).
 
 subscribe_to_queue(Config) ->
-    %% Test the node restart with both types of queues (quorum and classic) to
-    %% ensure there are no regressions
     Node = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
 
     Ch = rabbit_ct_client_helpers:open_channel(Config, Node),
@@ -337,8 +331,6 @@ subscribe_to_queue(Config) ->
     wait_for_messages(Config, QQ, <<"1">>, <<"1">>, <<"0">>).
 
 subscribe_with_autoack_to_queue(Config) ->
-    %% Test the node restart with both types of queues (quorum and classic) to
-    %% ensure there are no regressions
     Node = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
 
     Ch = rabbit_ct_client_helpers:open_channel(Config, Node),
@@ -362,6 +354,41 @@ subscribe_with_autoack_to_queue(Config) ->
     rabbit_ct_client_helpers:close_channel(Ch),
     wait_for_messages(Config, QQ, <<"0">>, <<"0">>, <<"0">>).
 
+consume_and_ack_from_queue(Config) ->
+    Node = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
+
+    Ch = rabbit_ct_client_helpers:open_channel(Config, Node),
+    QQ = <<"quorum-q">>,
+    ?assertEqual({'queue.declare_ok', QQ, 0, 0},
+                 declare(Ch, QQ, [{<<"x-queue-type">>, longstr, <<"quorum">>}])),
+
+    publish(Ch, QQ),
+    wait_for_messages(Config, QQ, <<"1">>, <<"1">>, <<"0">>),
+    %% TODO we don't store consumer tag for basic.get!!! could it be fixed?
+    DeliveryTag = consume(Ch, QQ, false),
+    wait_for_messages(Config, QQ, <<"1">>, <<"0">>, <<"1">>),
+    amqp_channel:cast(Ch, #'basic.ack'{delivery_tag = DeliveryTag}),
+    wait_for_messages(Config, QQ, <<"0">>, <<"0">>, <<"0">>).
+
+subscribe_and_ack_to_queue(Config) ->
+    Node = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
+
+    Ch = rabbit_ct_client_helpers:open_channel(Config, Node),
+    QQ = <<"quorum-q">>,
+    ?assertEqual({'queue.declare_ok', QQ, 0, 0},
+                 declare(Ch, QQ, [{<<"x-queue-type">>, longstr, <<"quorum">>}])),
+
+    publish(Ch, QQ),
+    wait_for_messages(Config, QQ, <<"1">>, <<"1">>, <<"0">>),
+    subscribe(Ch, QQ, false),
+    receive
+        {#'basic.deliver'{delivery_tag = DeliveryTag}, _} ->
+            ok
+    end,
+    wait_for_messages(Config, QQ, <<"1">>, <<"0">>, <<"1">>),
+    amqp_channel:cast(Ch, #'basic.ack'{delivery_tag = DeliveryTag}),
+    wait_for_messages(Config, QQ, <<"0">>, <<"0">>, <<"0">>).
+
 %%----------------------------------------------------------------------------
 
 declare(Ch, Q) ->
@@ -408,9 +435,10 @@ publish(Ch, Queue) ->
                                      payload = <<"msg">>}).
 
 consume(Ch, Queue, NoAck) ->
-    ?assertMatch({#'basic.get_ok'{}, #amqp_msg{payload = <<"msg">>}},
-                 amqp_channel:call(Ch, #'basic.get'{queue = Queue,
-                                                    no_ack = NoAck})).
+    {GetOk, _} = Reply = amqp_channel:call(Ch, #'basic.get'{queue = Queue,
+                                                            no_ack = NoAck}),
+    ?assertMatch({#'basic.get_ok'{}, #amqp_msg{payload = <<"msg">>}}, Reply),
+    GetOk#'basic.get_ok'.delivery_tag.
 
 consume_empty(Ch, Queue, NoAck) ->
     ?assertMatch(#'basic.get_empty'{},