Skip to content

Commit d06ddb9

Browse files
michaelklishinmichaelklishin
authored
Merge pull request #9576 from rabbitmq/mergify/bp/v3.12.x/pr-9572
Monitor members to the same stream in stream connection (backport #9572)
2 parents 70d07b6 + d1f597a commit d06ddb9

File tree

5 files changed

+132
-53
lines changed

5 files changed

+132
-53
lines changed

‎deps/rabbitmq_stream/src/rabbit_stream_reader.erl

Lines changed: 59 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -84,7 +84,7 @@
8484
heartbeat :: undefined | integer(),
8585
heartbeater :: any(),
8686
client_properties = #{} :: #{binary() => binary()},
87-
monitors = #{} :: #{reference() => stream()},
87+
monitors = #{} :: #{reference() => {pid(), stream()}},
8888
stats_timer :: undefined | rabbit_event:state(),
8989
resource_alarm :: boolean(),
9090
send_file_oct ::
@@ -916,10 +916,10 @@ open(info, {'DOWN', MonitorRef, process, _OsirisPid, _Reason},
916916
StatemData) ->
917917
{Connection1, State1} =
918918
case Monitors of
919-
#{MonitorRef := Stream} ->
919+
#{MonitorRef := {MemberPid, Stream}} ->
920920
Monitors1 = maps:remove(MonitorRef, Monitors),
921921
C = Connection#stream_connection{monitors = Monitors1},
922-
case clean_state_after_stream_deletion_or_failure(Stream, C,
922+
case clean_state_after_stream_deletion_or_failure(MemberPid, Stream, C,
923923
State)
924924
of
925925
{cleaned, NewConnection, NewState} ->
@@ -1852,7 +1852,7 @@ handle_frame_post_auth(Transport,
18521852
{request, CorrelationId,
18531853
{delete_publisher, PublisherId}}) ->
18541854
case Publishers of
1855-
#{PublisherId := #publisher{stream = Stream, reference = Ref}} ->
1855+
#{PublisherId := #publisher{stream = Stream, reference = Ref, leader = LeaderPid}} ->
18561856
Connection1 =
18571857
Connection0#stream_connection{publishers =
18581858
maps:remove(PublisherId,
@@ -1861,7 +1861,7 @@ handle_frame_post_auth(Transport,
18611861
maps:remove({Stream, Ref},
18621862
PubToIds)},
18631863
Connection2 =
1864-
maybe_clean_connection_from_stream(Stream, Connection1),
1864+
maybe_clean_connection_from_stream(LeaderPid, Stream, Connection1),
18651865
response(Transport,
18661866
Connection1,
18671867
delete_publisher,
@@ -2350,7 +2350,7 @@ handle_frame_post_auth(Transport,
23502350
CorrelationId),
23512351
{Connection1, State1} =
23522352
case
2353-
clean_state_after_stream_deletion_or_failure(Stream,
2353+
clean_state_after_stream_deletion_or_failure(undefined, Stream,
23542354
Connection,
23552355
State)
23562356
of
@@ -3088,7 +3088,7 @@ stream_r(Stream, #stream_connection{virtual_host = VHost}) ->
30883088
kind = queue,
30893089
virtual_host = VHost}.
30903090

3091-
clean_state_after_stream_deletion_or_failure(Stream,
3091+
clean_state_after_stream_deletion_or_failure(MemberPid, Stream,
30923092
#stream_connection{virtual_host =
30933093
VirtualHost,
30943094
stream_subscriptions
@@ -3113,16 +3113,30 @@ clean_state_after_stream_deletion_or_failure(Stream,
31133113
#{Stream := SubscriptionIds} = StreamSubscriptions,
31143114
Requests1 = lists:foldl(
31153115
fun(SubId, Rqsts0) ->
3116-
rabbit_stream_metrics:consumer_cancelled(self(),
3117-
stream_r(Stream,
3118-
C0),
3119-
SubId),
31203116
#{SubId := Consumer} = Consumers,
3121-
Rqsts1 = maybe_unregister_consumer(
3122-
VirtualHost, Consumer,
3123-
single_active_consumer(Consumer),
3124-
Rqsts0),
3125-
Rqsts1
3117+
case {MemberPid, Consumer} of
3118+
{undefined, _C} ->
3119+
rabbit_stream_metrics:consumer_cancelled(self(),
3120+
stream_r(Stream,
3121+
C0),
3122+
SubId),
3123+
maybe_unregister_consumer(
3124+
VirtualHost, Consumer,
3125+
single_active_consumer(Consumer),
3126+
Rqsts0);
3127+
{MemberPid, #consumer{configuration =
3128+
#consumer_configuration{member_pid = MemberPid}}} ->
3129+
rabbit_stream_metrics:consumer_cancelled(self(),
3130+
stream_r(Stream,
3131+
C0),
3132+
SubId),
3133+
maybe_unregister_consumer(
3134+
VirtualHost, Consumer,
3135+
single_active_consumer(Consumer),
3136+
Rqsts0);
3137+
_ ->
3138+
Rqsts0
3139+
end
31263140
end, Requests0, SubscriptionIds),
31273141
{true,
31283142
C0#stream_connection{stream_subscriptions =
@@ -3141,17 +3155,25 @@ clean_state_after_stream_deletion_or_failure(Stream,
31413155
{PurgedPubs, PurgedPubToIds} =
31423156
maps:fold(fun(PubId,
31433157
#publisher{stream = S, reference = Ref},
3144-
{Pubs, PubToIds}) ->
3145-
case S of
3146-
Stream ->
3158+
{Pubs, PubToIds}) when S =:= Stream andalso MemberPid =:= undefined ->
3159+
rabbit_stream_metrics:publisher_deleted(self(),
3160+
stream_r(Stream,
3161+
C1),
3162+
PubId),
3163+
{maps:remove(PubId, Pubs),
3164+
maps:remove({Stream, Ref}, PubToIds)};
3165+
(PubId,
3166+
#publisher{stream = S, reference = Ref, leader = MPid},
3167+
{Pubs, PubToIds}) when S =:= Stream andalso MPid =:= MemberPid ->
31473168
rabbit_stream_metrics:publisher_deleted(self(),
3148-
stream_r(S,
3169+
stream_r(Stream,
31493170
C1),
31503171
PubId),
31513172
{maps:remove(PubId, Pubs),
31523173
maps:remove({Stream, Ref}, PubToIds)};
3153-
_ -> {Pubs, PubToIds}
3154-
end
3174+
3175+
(_PubId, _Publisher, {Pubs, PubToIds}) ->
3176+
{Pubs, PubToIds}
31553177
end,
31563178
{Publishers, PublisherToIds}, Publishers),
31573179
{true,
@@ -3173,7 +3195,7 @@ clean_state_after_stream_deletion_or_failure(Stream,
31733195
orelse LeadersCleaned
31743196
of
31753197
true ->
3176-
C3 = demonitor_stream(Stream, C2),
3198+
C3 = demonitor_stream(MemberPid, Stream, C2),
31773199
{cleaned, C3#stream_connection{stream_leaders = Leaders1}, S2};
31783200
false ->
31793201
{not_cleaned, C2#stream_connection{stream_leaders = Leaders1}, S2}
@@ -3212,7 +3234,7 @@ remove_subscription(SubscriptionId,
32123234
#stream_connection_state{consumers = Consumers} = State) ->
32133235
#{SubscriptionId := Consumer} = Consumers,
32143236
#consumer{log = Log,
3215-
configuration = #consumer_configuration{stream = Stream}} =
3237+
configuration = #consumer_configuration{stream = Stream, member_pid = MemberPid}} =
32163238
Consumer,
32173239
rabbit_log:debug("Deleting subscription ~tp (stream ~tp)",
32183240
[SubscriptionId, Stream]),
@@ -3232,7 +3254,7 @@ remove_subscription(SubscriptionId,
32323254
Connection#stream_connection{stream_subscriptions =
32333255
StreamSubscriptions1},
32343256
Consumers1 = maps:remove(SubscriptionId, Consumers),
3235-
Connection2 = maybe_clean_connection_from_stream(Stream, Connection1),
3257+
Connection2 = maybe_clean_connection_from_stream(MemberPid, Stream, Connection1),
32363258
rabbit_stream_metrics:consumer_cancelled(self(),
32373259
stream_r(Stream, Connection2),
32383260
SubscriptionId),
@@ -3245,7 +3267,7 @@ remove_subscription(SubscriptionId,
32453267
{Connection2#stream_connection{outstanding_requests = Requests1},
32463268
State#stream_connection_state{consumers = Consumers1}}.
32473269

3248-
maybe_clean_connection_from_stream(Stream,
3270+
maybe_clean_connection_from_stream(MemberPid, Stream,
32493271
#stream_connection{stream_leaders =
32503272
Leaders} =
32513273
Connection0) ->
@@ -3254,7 +3276,7 @@ maybe_clean_connection_from_stream(Stream,
32543276
stream_has_subscriptions(Stream, Connection0)}
32553277
of
32563278
{false, false} ->
3257-
demonitor_stream(Stream, Connection0);
3279+
demonitor_stream(MemberPid, Stream, Connection0);
32583280
_ ->
32593281
Connection0
32603282
end,
@@ -3263,26 +3285,27 @@ maybe_clean_connection_from_stream(Stream,
32633285

32643286
maybe_monitor_stream(Pid, Stream,
32653287
#stream_connection{monitors = Monitors} = Connection) ->
3266-
case lists:member(Stream, maps:values(Monitors)) of
3288+
case lists:member({Pid, Stream}, maps:values(Monitors)) of
32673289
true ->
32683290
Connection;
32693291
false ->
32703292
MonitorRef = monitor(process, Pid),
32713293
Connection#stream_connection{monitors =
3272-
maps:put(MonitorRef, Stream,
3294+
maps:put(MonitorRef, {Pid, Stream},
32733295
Monitors)}
32743296
end.
32753297

3276-
demonitor_stream(Stream,
3298+
demonitor_stream(MemberPid, Stream,
32773299
#stream_connection{monitors = Monitors0} = Connection) ->
32783300
Monitors =
3279-
maps:fold(fun(MonitorRef, Strm, Acc) ->
3280-
case Strm of
3281-
Stream ->
3282-
demonitor(MonitorRef, [flush]),
3301+
maps:fold(fun(MonitorRef, {MPid, Strm}, Acc) when MPid =:= MemberPid andalso Strm =:= Stream ->
3302+
demonitor(MonitorRef, [flush]),
3303+
Acc;
3304+
(MonitorRef, {_MPid, Strm}, Acc) when MemberPid =:= undefined andalso Strm =:= Stream ->
3305+
demonitor(MonitorRef, [flush]),
32833306
Acc;
3284-
_ -> maps:put(MonitorRef, Strm, Acc)
3285-
end
3307+
(MonitorRef, {MPid, Strm}, Acc) ->
3308+
maps:put(MonitorRef, {MPid, Strm}, Acc)
32863309
end,
32873310
#{}, Monitors0),
32883311
Connection#stream_connection{monitors = Monitors}.

‎deps/rabbitmq_stream/test/rabbit_stream_SUITE.erl

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -91,7 +91,8 @@ init_per_group(cluster = Group, Config) ->
9191
{rmq_nodes_count, 3},
9292
{rmq_nodename_suffix, Group},
9393
{tcp_ports_base},
94-
{rabbitmq_ct_tls_verify, verify_none}
94+
{rabbitmq_ct_tls_verify, verify_none},
95+
{find_crashes, false} %% we kill stream members in some tests
9596
]),
9697
rabbit_ct_helpers:run_setup_steps(
9798
Config1,

0 commit comments

Comments
 (0)