Skip to content

Commit fae336a

Browse files
michaelklishinmichaelklishin
authored
Merge pull request #14145 from rabbitmq/qq-sac-return-bug
QQ: fix SAC activation bug
2 parents b6b3b19 + dd6fd0c commit fae336a

File tree

3 files changed

+173
-12
lines changed

3 files changed

+173
-12
lines changed

‎deps/rabbit/src/rabbit_fifo.erl‎

Lines changed: 25 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -317,7 +317,8 @@ apply(Meta, #modify{consumer_key = ConsumerKey,
317317
_ ->
318318
{State, ok}
319319
end;
320-
apply(#{index := Idx} = Meta,
320+
apply(#{index := Idx,
321+
machine_version := MacVer} = Meta,
321322
#requeue{consumer_key = ConsumerKey,
322323
msg_id = MsgId,
323324
index = OldIdx,
@@ -344,7 +345,13 @@ apply(#{index := Idx} = Meta,
344345
Messages),
345346
enqueue_count = EnqCount + 1},
346347
State2 = update_or_remove_con(Meta, ConsumerKey, Con, State1),
347-
checkout(Meta, State0, State2, []);
348+
{State3, Effects} = case MacVer >= 7 of
349+
true ->
350+
activate_next_consumer({State2, []});
351+
false ->
352+
{State2, []}
353+
end,
354+
checkout(Meta, State0, State3, Effects);
348355
_ ->
349356
{State00, ok, []}
350357
end;
@@ -923,15 +930,16 @@ get_checked_out(CKey, From, To, #?STATE{consumers = Consumers}) ->
923930
end.
924931

925932
-spec version() -> pos_integer().
926-
version() -> 6.
933+
version() -> 7.
927934

928935
which_module(0) -> rabbit_fifo_v0;
929936
which_module(1) -> rabbit_fifo_v1;
930937
which_module(2) -> rabbit_fifo_v3;
931938
which_module(3) -> rabbit_fifo_v3;
932939
which_module(4) -> ?MODULE;
933940
which_module(5) -> ?MODULE;
934-
which_module(6) -> ?MODULE.
941+
which_module(6) -> ?MODULE;
942+
which_module(7) -> ?MODULE.
935943

936944
-define(AUX, aux_v3).
937945

@@ -1747,8 +1755,8 @@ maybe_enqueue(RaftIdx, Ts, From, MsgSeqNo, RawMsg,
17471755
{duplicate, State0, Effects0}
17481756
end.
17491757

1750-
return(#{} = Meta, ConsumerKey, MsgIds, IncrDelCount, Anns,
1751-
Checked, Effects0, State0)
1758+
return(#{machine_version := MacVer} = Meta, ConsumerKey,
1759+
MsgIds, IncrDelCount, Anns, Checked, Effects0, State0)
17521760
when is_map(Anns) ->
17531761
%% We requeue in the same order as messages got returned by the client.
17541762
{State1, Effects1} =
@@ -1768,7 +1776,13 @@ return(#{} = Meta, ConsumerKey, MsgIds, IncrDelCount, Anns,
17681776
_ ->
17691777
State1
17701778
end,
1771-
checkout(Meta, State0, State2, Effects1).
1779+
{State3, Effects2} = case MacVer >= 7 of
1780+
true ->
1781+
activate_next_consumer({State2, Effects1});
1782+
false ->
1783+
{State2, Effects1}
1784+
end,
1785+
checkout(Meta, State0, State3, Effects2).
17721786

17731787
% used to process messages that are finished
17741788
complete(Meta, ConsumerKey, [MsgId],
@@ -2798,7 +2812,10 @@ convert(Meta, 4, To, State) ->
27982812
convert(Meta, 5, To, State);
27992813
convert(Meta, 5, To, State) ->
28002814
%% no conversion needed, this version only includes a logic change
2801-
convert(Meta, 6, To, State).
2815+
convert(Meta, 6, To, State);
2816+
convert(Meta, 6, To, State) ->
2817+
%% no conversion needed, this version only includes a logic change
2818+
convert(Meta, 7, To, State).
28022819

28032820
smallest_raft_index(#?STATE{messages = Messages,
28042821
ra_indexes = Indexes,

‎deps/rabbit/test/quorum_queue_SUITE.erl‎

Lines changed: 68 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -95,6 +95,8 @@ groups() ->
9595
format,
9696
add_member_2,
9797
single_active_consumer_priority_take_over,
98+
single_active_consumer_priority_take_over_return,
99+
single_active_consumer_priority_take_over_requeue,
98100
single_active_consumer_priority,
99101
force_shrink_member_to_current_member,
100102
force_all_queues_shrink_member_to_current_member,
@@ -1145,6 +1147,72 @@ single_active_consumer_priority_take_over(Config) ->
11451147
?DEFAULT_AWAIT),
11461148
ok.
11471149

1150+
single_active_consumer_priority_take_over_return(Config) ->
1151+
single_active_consumer_priority_take_over_base(20, Config).
1152+
1153+
single_active_consumer_priority_take_over_requeue(Config) ->
1154+
single_active_consumer_priority_take_over_base(-1, Config).
1155+
1156+
single_active_consumer_priority_take_over_base(DelLimit, Config) ->
1157+
check_quorum_queues_v4_compat(Config),
1158+
1159+
[Server0, Server1, _Server2] = Nodes =
1160+
rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
1161+
1162+
MinMacVers = lists:min([V || {ok, V} <-
1163+
erpc:multicall(Nodes, rabbit_fifo, version, [])]),
1164+
if MinMacVers < 7 ->
1165+
throw({skip, "single_active_consumer_priority_take_over_base needs a higher machine verison"});
1166+
true ->
1167+
ok
1168+
end,
1169+
1170+
Ch1 = rabbit_ct_client_helpers:open_channel(Config, Server0),
1171+
Ch2 = rabbit_ct_client_helpers:open_channel(Config, Server1),
1172+
QName = ?config(queue_name, Config),
1173+
Q1 = <<QName/binary, "_1">>,
1174+
RaNameQ1 = binary_to_atom(<<"%2F", "_", Q1/binary>>, utf8),
1175+
QueryFun = fun rabbit_fifo:query_single_active_consumer/1,
1176+
Args = [{<<"x-queue-type">>, longstr, <<"quorum">>},
1177+
{<<"x-delivery-limit">>, long, DelLimit},
1178+
{<<"x-single-active-consumer">>, bool, true}],
1179+
?assertEqual({'queue.declare_ok', Q1, 0, 0}, declare(Ch1, Q1, Args)),
1180+
ok = subscribe(Ch1, Q1, false, <<"ch1-ctag1">>, [{"x-priority", byte, 1}]),
1181+
?assertMatch({ok, {_, {value, {<<"ch1-ctag1">>, _}}}, _},
1182+
rpc:call(Server0, ra, local_query, [RaNameQ1, QueryFun])),
1183+
#'confirm.select_ok'{} = amqp_channel:call(Ch2, #'confirm.select'{}),
1184+
publish_confirm(Ch2, Q1),
1185+
%% higher priority consumer attaches
1186+
ok = subscribe(Ch2, Q1, false, <<"ch2-ctag1">>, [{"x-priority", byte, 3}]),
1187+
1188+
%% Q1 should still have Ch1 as consumer as it has pending messages
1189+
?assertMatch({ok, {_, {value, {<<"ch1-ctag1">>, _}}}, _},
1190+
rpc:call(Server0, ra, local_query,
1191+
[RaNameQ1, QueryFun])),
1192+
1193+
%% ack the message
1194+
receive
1195+
{#'basic.deliver'{consumer_tag = <<"ch1-ctag1">>,
1196+
delivery_tag = DeliveryTag}, _} ->
1197+
amqp_channel:cast(Ch1, #'basic.nack'{delivery_tag = DeliveryTag})
1198+
after ?TIMEOUT ->
1199+
flush(1),
1200+
exit(basic_deliver_timeout)
1201+
end,
1202+
1203+
?awaitMatch({ok, {_, {value, {<<"ch2-ctag1">>, _}}}, _},
1204+
rpc:call(Server0, ra, local_query, [RaNameQ1, QueryFun]),
1205+
?DEFAULT_AWAIT),
1206+
receive
1207+
{#'basic.deliver'{consumer_tag = <<"ch2-ctag1">>,
1208+
delivery_tag = DeliveryTag2}, _} ->
1209+
amqp_channel:cast(Ch1, #'basic.ack'{delivery_tag = DeliveryTag2})
1210+
after ?TIMEOUT ->
1211+
flush(1),
1212+
exit(basic_deliver_timeout_2)
1213+
end,
1214+
ok.
1215+
11481216
single_active_consumer_priority(Config) ->
11491217
check_quorum_queues_v4_compat(Config),
11501218
[Server0, Server1, Server2] =

‎deps/rabbit/test/rabbit_fifo_SUITE.erl‎

Lines changed: 80 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -42,12 +42,12 @@ groups() ->
4242
].
4343

4444
init_per_group(tests, Config) ->
45-
[{machine_version, 5} | Config];
45+
[{machine_version, rabbit_fifo:version()} | Config];
4646
init_per_group(machine_version_conversion, Config) ->
4747
Config.
4848

4949
init_per_testcase(_Testcase, Config) ->
50-
FF = ?config(machine_version, Config) == 5,
50+
FF = ?config(machine_version, Config) == rabbit_fifo:version(),
5151
ok = meck:new(rabbit_feature_flags, [passthrough]),
5252
meck:expect(rabbit_feature_flags, is_enabled, fun (_) -> FF end),
5353
Config.
@@ -1932,6 +1932,83 @@ single_active_consumer_higher_waiting_disconnected_test(Config) ->
19321932

19331933
ok.
19341934

1935+
single_active_consumer_higher_waiting_return_test(Config) ->
1936+
S0 = init(#{name => ?FUNCTION_NAME,
1937+
queue_resource => rabbit_misc:r("/", queue, ?FUNCTION_NAME_B),
1938+
single_active_consumer_on => true}),
1939+
1940+
Pid1 = test_util:fake_pid(node()),
1941+
C1Pid = test_util:fake_pid(n1@banana),
1942+
C2Pid = test_util:fake_pid(n2@banana),
1943+
% % adding some consumers
1944+
{CK1, C1} = {?LINE, {?LINE_B, C1Pid}},
1945+
{CK2, C2} = {?LINE, {?LINE_B, C2Pid}},
1946+
Entries =
1947+
[
1948+
%% add a consumer
1949+
{CK1, make_checkout(C1, {auto, {simple_prefetch, 1}}, #{priority => 1})},
1950+
?ASSERT(#rabbit_fifo{consumers = #{CK1 := #consumer{status = up}},
1951+
waiting_consumers = []}),
1952+
1953+
%% enqueue a message
1954+
{?LINE , rabbit_fifo:make_enqueue(Pid1, 1, msg1)},
1955+
1956+
%% add a consumer with a higher priority, current is quiescing
1957+
{CK2, make_checkout(C2, {auto, {simple_prefetch, 1}}, #{priority => 2})},
1958+
?ASSERT(#rabbit_fifo{consumers = #{CK1 := #consumer{status = quiescing}},
1959+
waiting_consumers = [{CK2, _}]}),
1960+
%% C1 returns message
1961+
{?LINE, rabbit_fifo:make_return(CK1, [0])},
1962+
%% C2 should activated
1963+
?ASSERT(#rabbit_fifo{consumers = #{CK2 := #consumer{status = up,
1964+
checked_out = Ch,
1965+
credit = 0}},
1966+
waiting_consumers = [_]} when map_size(Ch) == 1)
1967+
],
1968+
{_S1, _} = run_log(Config, S0, Entries, fun single_active_invariant/1),
1969+
1970+
ok.
1971+
1972+
single_active_consumer_higher_waiting_requeue_test(Config) ->
1973+
S0 = init(#{name => ?FUNCTION_NAME,
1974+
queue_resource => rabbit_misc:r("/", queue, ?FUNCTION_NAME_B),
1975+
single_active_consumer_on => true}),
1976+
1977+
Pid1 = test_util:fake_pid(node()),
1978+
C1Pid = test_util:fake_pid(n1@banana),
1979+
C2Pid = test_util:fake_pid(n2@banana),
1980+
% % adding some consumers
1981+
{CK1, C1} = {?LINE, {?LINE_B, C1Pid}},
1982+
EnqIdx = ?LINE,
1983+
RequeueIdx = ?LINE,
1984+
{CK2, C2} = {?LINE, {?LINE_B, C2Pid}},
1985+
Entries =
1986+
[
1987+
%% add a consumer
1988+
{CK1, make_checkout(C1, {auto, {simple_prefetch, 1}}, #{priority => 1})},
1989+
?ASSERT(#rabbit_fifo{consumers = #{CK1 := #consumer{status = up}},
1990+
waiting_consumers = []}),
1991+
1992+
%% enqueue a message
1993+
{EnqIdx , rabbit_fifo:make_enqueue(Pid1, 1, msg1)},
1994+
1995+
%% add a consumer with a higher priority, current is quiescing
1996+
{CK2, make_checkout(C2, {auto, {simple_prefetch, 1}}, #{priority => 2})},
1997+
?ASSERT(#rabbit_fifo{consumers = #{CK1 := #consumer{status = quiescing}},
1998+
waiting_consumers = [{CK2, _}]}),
1999+
%% C1 returns message
2000+
% {?LINE, rabbit_fifo:make_requeue(CK1, [0])},
2001+
{RequeueIdx , element(2, hd(rabbit_fifo:make_requeue(CK1, {notify, 1, self()},
2002+
[{0, EnqIdx, 0, msg1}], [])))},
2003+
%% C2 should activated
2004+
?ASSERT(#rabbit_fifo{consumers = #{CK2 := #consumer{status = up,
2005+
checked_out = Ch,
2006+
credit = 0}},
2007+
waiting_consumers = [_]} when map_size(Ch) == 1)
2008+
],
2009+
{_S1, _} = run_log(Config, S0, Entries, fun single_active_invariant/1),
2010+
2011+
ok.
19352012
single_active_consumer_quiescing_disconnected_test(Config) ->
19362013
S0 = init(#{name => ?FUNCTION_NAME,
19372014
queue_resource => rabbit_misc:r("/", queue, ?FUNCTION_NAME_B),
@@ -2455,8 +2532,7 @@ machine_version_test(C) ->
24552532
consumers = #{Cid := #consumer{cfg = #consumer_cfg{priority = 0}}},
24562533
service_queue = S,
24572534
messages = Msgs}, ok,
2458-
[_|_]} = apply(meta(C, Idx),
2459-
{machine_version, 0, 2}, S1),
2535+
[_|_]} = apply(meta(C, Idx), {machine_version, 0, 2}, S1),
24602536
%% validate message conversion to lqueue
24612537
?assertEqual(1, lqueue:len(Msgs)),
24622538
?assert(priority_queue:is_queue(S)),

0 commit comments

Comments
 (0)