Merge pull request #7849 from rabbitmq/mergify/bp/v3.12.x/pr-7827

Quorum queues: avoid potential crash when returning message. (backport #7827)

Author: michaelklishin

MODULE.bazel

@@ -267,8 +267,8 @@ erlang_package.hex_package(
 erlang_package.hex_package(
     name = "ra",
     build_file = "@rabbitmq-server//bazel:BUILD.ra",
-    sha256 = "321353d6f25bf3ac629698999a5d402308d21bf236bd68a6f4e92bee095f2e5f",
-    version = "2.5.0",
+    sha256 = "13b03f02cf6c1837c527edd4a953f0c09da0abad0af6985b64bfd66943c4c5c3",
+    version = "2.5.1",
 )
 
 erlang_package.hex_package(

deps/rabbit/src/rabbit_fifo.erl

@@ -1003,8 +1003,8 @@ handle_aux(leader, _, garbage_collection, Aux, Log, MacState) ->
     {no_reply, force_eval_gc(Log, MacState, Aux), Log};
 handle_aux(follower, _, garbage_collection, Aux, Log, MacState) ->
     {no_reply, force_eval_gc(Log, MacState, Aux), Log};
-handle_aux(leader, cast, {#return{msg_ids = MsgIds,
-                                  consumer_id = ConsumerId}, Corr, Pid},
+handle_aux(_RaftState, cast, {#return{msg_ids = MsgIds,
+                                      consumer_id = ConsumerId}, Corr, Pid},
            Aux0, Log0, #?MODULE{cfg = #cfg{delivery_limit = undefined},
                                 consumers = Consumers}) ->
     case Consumers of

deps/rabbit/test/quorum_queue_SUITE.erl

@@ -69,6 +69,7 @@ groups() ->
                                             metrics_cleanup_on_leadership_takeover,
                                             metrics_cleanup_on_leader_crash,
                                             consume_in_minority,
+                                            reject_after_leader_transfer,
                                             shrink_all,
                                             rebalance,
                                             file_handle_reservations,
@@ -907,6 +908,36 @@ consume_in_minority(Config) ->
     ok = rabbit_ct_broker_helpers:start_node(Config, Server2),
     ok.
 
+reject_after_leader_transfer(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),
+    RaName = binary_to_atom(<<"%2F_", QQ/binary>>, utf8),
+    ?assertEqual({'queue.declare_ok', QQ, 0, 0},
+                 declare(Ch, QQ, [{<<"x-queue-type">>, longstr, <<"quorum">>}])),
+    publish(Ch, QQ),
+
+    Ch2 = rabbit_ct_client_helpers:open_channel(Config, Server2),
+    {#'basic.get_ok'{delivery_tag = Tag}, #amqp_msg{}} =
+        basic_get(Ch2, QQ, false, 10),
+
+    ServerId1 = {RaName, Server1},
+    ct:pal("transfer leadership ~p",
+           [rabbit_ct_broker_helpers:rpc(Config, 0, ra,
+                                         transfer_leadership, [ServerId1, ServerId1])]),
+    ok = amqp_channel:call(Ch2, #'basic.reject'{delivery_tag = Tag,
+                                                requeue = true}),
+    wait_for_messages(Config, [[QQ, <<"1">>, <<"1">>, <<"0">>]]),
+
+    {#'basic.get_ok'{delivery_tag = Tag2}, #amqp_msg{}} =
+        basic_get(Ch2, QQ, false, 10),
+
+    ok = amqp_channel:call(Ch2, #'basic.reject'{delivery_tag = Tag2,
+                                                requeue = true}),
+    ok.
+
 shrink_all(Config) ->
     [Server0, Server1, Server2] =
         rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
@@ -3340,3 +3371,14 @@ validate_queue(Ch, Queue, ExpectedMsgs) ->
          end
      end || M <- ExpectedMsgs],
     ok.
+
+basic_get(_, _, _, 0) ->
+    empty;
+basic_get(Ch, Q, NoAck, Attempt) ->
+    case amqp_channel:call(Ch, #'basic.get'{queue = Q, no_ack = NoAck}) of
+        {#'basic.get_ok'{}, #amqp_msg{}} = R ->
+            R;
+        _ ->
+            timer:sleep(100),
+            basic_get(Ch, Q, NoAck, Attempt - 1)
+    end.

rabbitmq-components.mk

@@ -115,7 +115,7 @@ dep_cowboy = hex 2.8.0
 dep_cowlib = hex 2.12.1
 dep_looking_glass = git https://github.com/rabbitmq/looking_glass.git master
 dep_prometheus = hex 4.10.0
-dep_ra = hex 2.5.0
+dep_ra = hex 2.5.1
 dep_ranch = hex 2.1.0
 dep_recon = hex 2.5.3
 dep_redbug = hex 2.0.7