Skip to content

Make consumer-level prefetch-like limit configurable for cases with QoS prefetch > 200 #11822

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 5 commits into from
Closed

Make consumer-level prefetch-like limit configurable for cases with QoS prefetch > 200 #11822

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
b5ea618
Configurable rabbit.classic_queue_consumer_unsent_message_limit
Ayanda-D Jul 24, 2024
708d071
Fix CQ consumer_unsent_message_limit config format
Ayanda-D Jul 25, 2024
e3ce46b
Store/retrieve UNSENT_MESSAGE_LIMIT using pesistent_term
Ayanda-D Jul 25, 2024
dcaeb67
Use default classic_queue_consumer_unsent_message_limit from Makefile
Ayanda-D Jul 29, 2024
45c4aa6
Add classic_queue_consumer_unsent_message_limit to BUILD.bazel
Ayanda-D Jul 29, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions deps/rabbit/BUILD.bazel
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -126,6 +126,7 @@ _APP_ENV = """[
{vhost_restart_strategy, continue},
%% {global, prefetch count}
{default_consumer_prefetch, {false, 0}},
{classic_queue_consumer_unsent_message_limit, 200},
%% interval at which the channel can perform periodic actions
{channel_tick_interval, 60000},
%% Default max message size is 16 MB
Expand Down
1 change: 1 addition & 0 deletions deps/rabbit/Makefile
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -109,6 +109,7 @@ define PROJECT_ENV
{vhost_restart_strategy, continue},
%% {global, prefetch count}
{default_consumer_prefetch, {false, 0}},
{classic_queue_consumer_unsent_message_limit, 200},
%% interval at which the channel can perform periodic actions
{channel_tick_interval, 60000},
%% Default max message size is 16 MB
Expand Down
10 changes: 10 additions & 0 deletions deps/rabbit/priv/schema/rabbit.schema
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1591,6 +1591,16 @@ end}.
{validators, ["non_zero_positive_integer"]}
]}.

%% Unsent Message Limit for classic queue consumer channels upon which
%% the channel blocked.
%%
%% {classic_queue_consumer_unsent_message_limit, 200},

{mapping, "classic_queue.consumer_unsent_message_limit", "rabbit.classic_queue_consumer_unsent_message_limit", [
{datatype, integer},
{validators, ["non_zero_positive_integer"]}
]}.

%% Product name & version overrides.

{mapping, "product.name", "rabbit.product_name", [
Expand Down
4 changes: 3 additions & 1 deletion deps/rabbit/src/rabbit_queue_consumers.erl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@

-define(QUEUE, lqueue).

-define(UNSENT_MESSAGE_LIMIT, 200).
-define(UNSENT_MESSAGE_LIMIT, persistent_term:get(unsent_message_limit)).

%% Utilisation average calculations are all in μs.
-define(USE_AVG_HALF_LIFE, 1000000.0).
Expand Down Expand Up @@ -168,6 +168,8 @@ add(ChPid, CTag, NoAck, LimiterPid, LimiterActive,
delivery_count = InitialDeliveryCount}}}
end,
update_ch_record(C),
{ok, UnsentMsgLimit} = application:get_env(rabbit, classic_queue_consumer_unsent_message_limit),
persistent_term:put(unsent_message_limit, UnsentMsgLimit),
Consumer = #consumer{tag = CTag,
ack_required = not NoAck,
prefetch = parse_prefetch_count(ModeOrPrefetch),
Expand Down
7 changes: 7 additions & 0 deletions deps/rabbit/test/config_schema_SUITE_data/rabbit.snippets
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -858,6 +858,13 @@ credential_validator.regexp = ^abc\\d+",
]}],
[]},

{rabbit_classic_queue_consumer_unsent_message_limit,
"classic_queue.consumer_unsent_message_limit = 200",
[{rabbit, [
{classic_queue_consumer_unsent_message_limit, 200}
]}],
[]},

{rabbit_msg_store_shutdown_timeout,
"message_store_shutdown_timeout = 600000",
[{rabbit, [
Expand Down
Loading