Quorum queues: avoid potential crash when returning message.

Returns reaching a Ra member that used to be leader but now has stepped
down would cause that follower to crash and restart.

This commit avoids this scenario as well as giving the return commands
a good chance of being resent to the new leader in a timeley manner.
(see the Ra release for this).

(cherry picked from commit 01f6d0f)

Author: kjnilsson

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 = "f5479a68c66bfee4ce96dcca81e4457cc99e4ab06bd350b83a0e5ac2ff29c78a",
+    version = "2.5.1-pre.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{}} =
+        amqp_channel:call(Ch2, #'basic.get'{queue = QQ, no_ack = false}),
+
+    ServerId1 = {RaName, Server1},
+    ct:pal("transfser 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{}} =
+        amqp_channel:call(Ch2, #'basic.get'{queue = QQ, no_ack = false}),
+
+    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),

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-pre.1
 dep_ranch = hex 2.1.0
 dep_recon = hex 2.5.3
 dep_redbug = hex 2.0.7