diff --git a/deps/rabbit/src/rabbit_fifo_dlx.erl b/deps/rabbit/src/rabbit_fifo_dlx.erl index 46402b5f4c2c..bbf5ffc60941 100644 --- a/deps/rabbit/src/rabbit_fifo_dlx.erl +++ b/deps/rabbit/src/rabbit_fifo_dlx.erl @@ -114,11 +114,20 @@ apply(_, {dlx, #checkout{consumer = Pid, apply(_, {dlx, #checkout{consumer = ConsumerPid, prefetch = Prefetch}}, at_least_once, - #?MODULE{consumer = #dlx_consumer{checked_out = CheckedOutOldConsumer}, + #?MODULE{consumer = #dlx_consumer{pid = OldConsumerPid, + checked_out = CheckedOutOldConsumer}, discards = Discards0, msg_bytes = Bytes, msg_bytes_checkout = BytesCheckout} = State0) -> %% Since we allow only a single consumer, the new consumer replaces the old consumer. + case ConsumerPid of + OldConsumerPid -> + ok; + _ -> + rabbit_log:debug("Terminating ~p since ~p becomes active rabbit_fifo_dlx_worker", + [OldConsumerPid, ConsumerPid]), + ensure_worker_terminated(State0) + end, %% All checked out messages to the old consumer need to be returned to the discards queue %% such that these messages will be re-delivered to the new consumer. %% When inserting back into the discards queue, we respect the original order in which messages diff --git a/deps/rabbit/src/rabbit_fifo_dlx_client.erl b/deps/rabbit/src/rabbit_fifo_dlx_client.erl index 00e94276061a..f5014efb0427 100644 --- a/deps/rabbit/src/rabbit_fifo_dlx_client.erl +++ b/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 ~p on quorum queue leader ~p because actual leader is ~p.", - [Cmd, Leader, L]), - {error, ra_command_failed}; + [Cmd, Leader, NonLocalLeader]), + {error, non_local_leader}; Err -> rabbit_log:warning("Failed to process command ~p on quorum queue leader ~p: ~p~n" "Trying ~b more time(s)...", diff --git a/deps/rabbit/src/rabbit_fifo_dlx_sup.erl b/deps/rabbit/src/rabbit_fifo_dlx_sup.erl index f72040f8f2f5..af7621f3f9b4 100644 --- a/deps/rabbit/src/rabbit_fifo_dlx_sup.erl +++ b/deps/rabbit/src/rabbit_fifo_dlx_sup.erl @@ -24,11 +24,12 @@ start_link() -> init([]) -> SupFlags = #{strategy => simple_one_for_one, - intensity => 1, - period => 5}, + intensity => 100, + period => 1}, Worker = rabbit_fifo_dlx_worker, ChildSpec = #{id => Worker, start => {Worker, start_link, []}, type => worker, + restart => transient, modules => [Worker]}, {ok, {SupFlags, [ChildSpec]}}. diff --git a/deps/rabbit/src/rabbit_fifo_dlx_worker.erl b/deps/rabbit/src/rabbit_fifo_dlx_worker.erl index 04b8974007d6..3d5c49fe23a8 100644 --- a/deps/rabbit/src/rabbit_fifo_dlx_worker.erl +++ b/deps/rabbit/src/rabbit_fifo_dlx_worker.erl @@ -106,7 +106,7 @@ 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), @@ -114,15 +114,19 @@ handle_continue(QRef, undefined) -> 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).