Do not restart DLX worker if leader is non-local

The rabbit_fifo_dlx_worker should be co-located with the quorum
queue leader.

If a new leader on a different node gets elected before the
rabbit_fifo_dlx_worker initialises (i.e. registers itself as a
consumer), it should stop itself normally, such that it is not restarted
by rabbit_fifo_dlx_sup.

Another rabbit_fifo_dlx_worker should be created on the new quorum
queue leader node.

Author: ansd

deps/rabbit/src/rabbit_fifo_dlx_client.erl

@@ -32,7 +32,7 @@ settle(MsgIds, #state{leader = Leader} = State)
     {ok, State}.
 
 -spec checkout(rabbit_amqqueue:name(), ra:server_id(), non_neg_integer()) ->
-    {ok, state()} | {error, ra_command_failed}.
+    {ok, state()} | {error, non_local_leader | ra_command_failed}.
 checkout(QResource, Leader, NumUnsettled) ->
     Cmd = rabbit_fifo_dlx:make_checkout(self(), NumUnsettled),
     State = #state{queue_resource = QResource,
@@ -46,10 +46,10 @@ process_command(Cmd, #state{leader = Leader} = State, Tries) ->
     case ra:process_command(Leader, Cmd, 60_000) of
         {ok, ok, Leader} ->
             {ok, State#state{leader = Leader}};
-        {ok, ok, L} ->
+        {ok, ok, NonLocalLeader} ->
             rabbit_log:warning("Failed to process command ~tp on quorum queue leader ~tp because actual leader is ~tp.",
-                               [Cmd, Leader, L]),
-            {error, ra_command_failed};
+                               [Cmd, Leader, NonLocalLeader]),
+            {error, non_local_leader};
         Err ->
             rabbit_log:warning("Failed to process command ~tp on quorum queue leader ~tp: ~tp~n"
                                "Trying ~b more time(s)...",

deps/rabbit/src/rabbit_fifo_dlx_sup.erl

@@ -30,5 +30,6 @@ init([]) ->
     ChildSpec = #{id => Worker,
                   start => {Worker, start_link, []},
                   type => worker,
+                  restart => transient,
                   modules => [Worker]},
     {ok, {SupFlags, [ChildSpec]}}.

deps/rabbit/src/rabbit_fifo_dlx_worker.erl

@@ -106,23 +106,27 @@ init(QRef) ->
     {ok, undefined, {continue, QRef}}.
 
 -spec handle_continue(rabbit_amqqueue:name(), undefined) ->
-    {noreply, state()}.
+    {noreply, state()} | {stop, term(), undefined}.
 handle_continue(QRef, undefined) ->
     {ok, Prefetch} = application:get_env(rabbit,
                                          dead_letter_worker_consumer_prefetch),
     {ok, SettleTimeout} = application:get_env(rabbit,
                                               dead_letter_worker_publisher_confirm_timeout),
     {ok, Q} = rabbit_amqqueue:lookup(QRef),
     {ClusterName, _MaybeOldLeaderNode} = amqqueue:get_pid(Q),
-    {ok, ConsumerState} = rabbit_fifo_dlx_client:checkout(QRef,
-                                                          {ClusterName, node()},
-                                                          Prefetch),
-    {noreply, lookup_topology(#state{queue_ref = QRef,
-                                     queue_type_state = rabbit_queue_type:init(),
-                                     settle_timeout = SettleTimeout,
-                                     dlx_client_state = ConsumerState,
-                                     monitor_ref = erlang:monitor(process, ClusterName)
-                                    })}.
+    case rabbit_fifo_dlx_client:checkout(QRef, {ClusterName, node()}, Prefetch) of
+        {ok, ConsumerState} ->
+            {noreply, lookup_topology(#state{queue_ref = QRef,
+                                             queue_type_state = rabbit_queue_type:init(),
+                                             settle_timeout = SettleTimeout,
+                                             dlx_client_state = ConsumerState,
+                                             monitor_ref = erlang:monitor(process, ClusterName)
+                                            })};
+        {error, non_local_leader = Reason} ->
+            {stop, {shutdown, Reason}, undefined};
+        Error ->
+            {stop, Error, undefined}
+    end.
 
 terminate(_Reason, State) ->
     cancel_timer(State).