root
/
rabbitmq-server
mirror of https://github.com/rabbitmq/rabbitmq-server.git
1
0
Fork 0
Code Issues Actions 6 Packages Projects Releases Wiki Activity

Local shovels: fix handling of acks/nacks

This commit is contained in:
Diana Parra Corbacho 2025-08-28 12:29:20 +02:00
parent 7f1febe70b
commit 6e2e19591a
1 changed files with 12 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
deps/rabbitmq_shovel/src/rabbit_local_shovel.erl vendored
View File

@ -591,22 +591,20 @@ get_user_vhost_from_amqp_param(Uri) ->
settle(Op, DeliveryTag, Multiple,
#{source := #{queue_r := QRef,
current := Current = #{consumer_tag := CTag,
unacked_message_q := UAMQ0}
unacked_message_q := UAMQ0,
queue_states := QState0}
} = Src} = State0) ->
{MsgIds, UAMQ} = collect_acks(UAMQ0, DeliveryTag, Multiple),
State = State0#{source => Src#{current => Current#{unacked_message_q => UAMQ}}},
lists:foldl(
fun(MsgId, #{source := Src0 = #{current := Current0 = #{queue_states := QState0}}} = St0) ->
case rabbit_queue_type:settle(QRef, Op, CTag, [MsgId], QState0) of
case rabbit_queue_type:settle(QRef, Op, CTag, lists:reverse(MsgIds), QState0) of
{ok, QState1, Actions} ->
St = St0#{source => Src0#{current => Current0#{queue_states => QState1}}},
handle_queue_actions(Actions, St);
State = State0#{source => Src#{current => Current#{queue_states => QState1,
unacked_message_q => UAMQ}}},
handle_queue_actions(Actions, State);
{'protocol_error', Type, Reason, Args} ->
?LOG_ERROR("Shovel failed to settle ~p acknowledgments with ~tp: ~tp",
[Op, Type, io_lib:format(Reason, Args)]),
exit({shutdown, {ack_failed, Reason}})
end
end, State, MsgIds).
end.
%% From rabbit_channel
%% Records a client-sent acknowledgement. Handles both single delivery acks