Support for publisher confirms in quorum queues

[#154472198]

Author: dcorbacho

src/rabbit_amqqueue.erl

@@ -1304,7 +1304,8 @@ deliver([], _Delivery) ->
     %% /dev/null optimisation
     [];
 
-deliver(Qs, Delivery = #delivery{flow = Flow}) ->
+deliver(Qs, Delivery = #delivery{flow = Flow,
+                                 confirm = Confirm}) ->
     {Quorum, MPids, SPids} = qpids(Qs),
     QPids = MPids ++ SPids,
     %% We use up two credits to send to a slave since the message
@@ -1331,7 +1332,13 @@ deliver(Qs, Delivery = #delivery{flow = Flow}) ->
     SMsg = {deliver, Delivery, true},
     delegate:invoke_no_result(MPids, {gen_server2, cast, [MMsg]}),
     delegate:invoke_no_result(SPids, {gen_server2, cast, [SMsg]}),
-    [ra:send(Q, {enqueue, Delivery#delivery.message}) || Q <- Quorum],
+    case Confirm of
+        false ->
+            [ra:send(Q, {enqueue, Delivery#delivery.message}) || Q <- Quorum];
+        true ->
+            [ra:send_and_notify(Q, {enqueue, Delivery#delivery.message},
+                                Delivery#delivery.msg_seq_no) || Q <- Quorum]
+    end,
     QPids ++ Quorum.
 
 qpids([]) -> {[], [], []}; %% optimisation

src/rabbit_channel.erl

@@ -596,10 +596,8 @@ handle_cast({reject_publish, MsgSeqNo, _QPid}, State = #ch{unconfirmed = UC}) ->
     %% NB: don't call noreply/1 since we don't want to send confirms.
     noreply_coalesce(record_rejects(MXs, State#ch{unconfirmed = UC1}));
 
-handle_cast({confirm, MsgSeqNos, QPid}, State = #ch{unconfirmed = UC}) ->
-    {MXs, UC1} = dtree:take(MsgSeqNos, QPid, UC),
-    %% NB: don't call noreply/1 since we don't want to send confirms.
-    noreply_coalesce(record_confirms(MXs, State#ch{unconfirmed = UC1})).
+handle_cast({confirm, MsgSeqNos, QPid}, State) ->
+    noreply_coalesce(confirm(MsgSeqNos, QPid, State)).
 
 handle_info({ra_fifo, {RegisteredName, _} = Id, {delivery, CTag, MsgId, Msg}},
             #ch{consumer_mapping = ConsumerMapping} = State) ->
@@ -618,6 +616,9 @@ handle_info({ra_fifo, {RegisteredName, _} = Id, {delivery, CTag, MsgId, Msg}},
     noreply(handle_deliver(CTag, AckRequired,
                            {RegisteredName, Id, MsgId, true, Msg}, State));
 
+handle_info({ra_event, {applied, Id, MsgSeqNo}} = Evt, State) ->
+    noreply_coalesce(confirm([MsgSeqNo], Id, State));
+
 handle_info({bump_credit, Msg}, State) ->
     %% A rabbit_amqqueue_process is granting credit to our channel. If
     %% our channel was being blocked by this process, and no other
@@ -1888,6 +1889,11 @@ process_routing_confirm(true, QPids,  MsgSeqNo,  XName, State) ->
     State#ch{unconfirmed = dtree:insert(MsgSeqNo, QPids, XName,
                                         State#ch.unconfirmed)}.
 
+confirm(MsgSeqNos, QPid, State = #ch{unconfirmed = UC}) ->
+    {MXs, UC1} = dtree:take(MsgSeqNos, QPid, UC),
+    %% NB: don't call noreply/1 since we don't want to send confirms.
+    record_confirms(MXs, State#ch{unconfirmed = UC1}).
+
 send_nacks([], State) ->
     State;
 send_nacks(_MXs, State = #ch{state = closing,

test/quorum_queue_SUITE.erl

@@ -46,7 +46,8 @@ groups() ->
           consume_and_ack,
           subscribe_and_ack,
           consume_and_single_nack,
-          subscribe_and_single_nack
+          subscribe_and_single_nack,
+          publisher_confirms
         ]}
     ].
 
@@ -242,8 +243,6 @@ stop_start_rabbit_app(Config) ->
         amqp_channel:call(Ch2, #'basic.get'{queue  = CQ2, no_ack = false}).
 
 publish(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),
@@ -430,6 +429,20 @@ subscribe_and_single_nack(Config) ->
                                         requeue      = true}),
     wait_for_messages(Config, QQ, <<"1">>, <<"1">>, <<"0">>).
 
+publisher_confirms(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">>}])),
+
+    amqp_channel:call(Ch, #'confirm.select'{}),
+    amqp_channel:register_confirm_handler(Ch, self()),
+    publish(Ch, QQ),
+    wait_for_messages(Config, QQ, <<"1">>, <<"1">>, <<"0">>),
+    amqp_channel:wait_for_confirms(Ch).
+
 %%----------------------------------------------------------------------------
 
 declare(Ch, Q) ->