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

Merge pull request #14146 from rabbitmq/mergify/bp/v4.1.x/pr-14143
Trigger a 4.1.x alpha release build / trigger_alpha_build (push) Waiting to run Details
Test (make) / Build and Xref (1.17, 26) (push) Waiting to run Details
Test (make) / Build and Xref (1.17, 27) (push) Waiting to run Details
Test (make) / Test (1.17, 27, khepri) (push) Waiting to run Details
Test (make) / Test (1.17, 27, mnesia) (push) Waiting to run Details
Test (make) / Test mixed clusters (1.17, 27, khepri) (push) Waiting to run Details
Test (make) / Test mixed clusters (1.17, 27, mnesia) (push) Waiting to run Details
Test (make) / Type check (1.17, 27) (push) Waiting to run Details 

Close stream consumer log after stream is deleted or unavailable (backport #14143)
This commit is contained in:
Arnaud Cogoluègnes 2025-06-27 11:53:42 +00:00 committed by GitHub
parent 7d9dbf98a6 8b64572820
commit 17942ec9f6
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 7 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
deps/rabbitmq_stream/src/rabbit_stream_reader.erl vendored
View File

@ -3309,25 +3309,29 @@ clean_subscriptions(MemberPid, Stream,
fun(SubId, {DelSubIds, Rqsts0}) -> fun(SubId, {DelSubIds, Rqsts0}) ->
#{SubId := Consumer} = Consumers, #{SubId := Consumer} = Consumers,
case {MemberPid, Consumer} of case {MemberPid, Consumer} of
{undefined, _C} -> {undefined, #consumer{log = Log}} ->
rabbit_stream_metrics:consumer_cancelled(self(), rabbit_stream_metrics:consumer_cancelled(self(),
stream_r(Stream, stream_r(Stream,
C0), C0),
SubId, SubId,
Username), Username),
close_log(Log),
Rqsts1 = maybe_unregister_consumer( Rqsts1 = maybe_unregister_consumer(
VirtualHost, Consumer, VirtualHost, Consumer,
single_active_consumer(Consumer), single_active_consumer(Consumer),
Rqsts0), Rqsts0),
{[SubId | DelSubIds], Rqsts1}; {[SubId | DelSubIds], Rqsts1};
{MemberPid, {MemberPid,
#consumer{configuration = #consumer{
#consumer_configuration{member_pid = MemberPid}}} -> log = Log,
configuration = #consumer_configuration{member_pid = MemberPid}}} ->
rabbit_stream_metrics:consumer_cancelled(self(), rabbit_stream_metrics:consumer_cancelled(self(),
stream_r(Stream, stream_r(Stream,
C0), C0),
SubId, SubId,
Username), Username),
close_log(Log),
Rqsts1 = maybe_unregister_consumer( Rqsts1 = maybe_unregister_consumer(
VirtualHost, Consumer, VirtualHost, Consumer,
single_active_consumer(Consumer), single_active_consumer(Consumer),