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

Merge pull request #5085 from rabbitmq/amqp-stream-consumer-file-handle-leak-fix

This commit is contained in:
Michael Klishin 2022-06-21 23:17:23 +04:00 committed by GitHub
parent 2e3ba4c1d7 8635ded7fe
commit bebe9654a0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 18 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
deps/rabbit/src/rabbit_stream_queue.erl vendored
View File

@ -312,14 +312,20 @@ begin_stream(#stream_client{name = QName, readers = Readers0} = State0,
cancel(_Q, ConsumerTag, OkMsg, ActingUser, #stream_client{readers = Readers0,
name = QName} = State) ->
Readers = maps:remove(ConsumerTag, Readers0),
rabbit_core_metrics:consumer_deleted(self(), ConsumerTag, QName),
rabbit_event:notify(consumer_deleted, [{consumer_tag, ConsumerTag},
{channel, self()},
{queue, QName},
{user_who_performed_action, ActingUser}]),
maybe_send_reply(self(), OkMsg),
{ok, State#stream_client{readers = Readers}}.
case maps:take(ConsumerTag, Readers0) of
{#stream{log = Log}, Readers} ->
ok = close_log(Log),
rabbit_core_metrics:consumer_deleted(self(), ConsumerTag, QName),
rabbit_event:notify(consumer_deleted,
[{consumer_tag, ConsumerTag},
{channel, self()},
{queue, QName},
{user_who_performed_action, ActingUser}]),
maybe_send_reply(self(), OkMsg),
{ok, State#stream_client{readers = Readers}};
error ->
{ok, State}
end.
credit(CTag, Credit, Drain, #stream_client{readers = Readers0,
name = Name,
@ -1019,3 +1025,7 @@ set_leader_pid(Pid, QName) ->
_ ->
ok
end.
close_log(undefined) -> ok;
close_log(Log) ->
osiris_log:close(Log).