diff --git a/deps/rabbit/priv/schema/rabbit.schema b/deps/rabbit/priv/schema/rabbit.schema index 1b4d43593ce..c589f21a61f 100644 --- a/deps/rabbit/priv/schema/rabbit.schema +++ b/deps/rabbit/priv/schema/rabbit.schema @@ -2779,6 +2779,11 @@ end}. {mapping, "stream.read_ahead", "rabbit.stream_read_ahead", [{datatype, {enum, [true, false]}}]}. +{mapping, "stream.read_ahead_limit", "rabbit.stream_read_ahead_limit", [ + {datatype, [integer, string]}, + {validators, ["is_supported_information_unit"]} +]}. + {mapping, "cluster_tags.$tag", "rabbit.cluster_tags", [ {datatype, [binary]} ]}. diff --git a/deps/rabbit/src/rabbit_stream_queue.erl b/deps/rabbit/src/rabbit_stream_queue.erl index affe36296df..95f8d6b2266 100644 --- a/deps/rabbit/src/rabbit_stream_queue.erl +++ b/deps/rabbit/src/rabbit_stream_queue.erl @@ -49,7 +49,7 @@ -export([format_osiris_event/2]). -export([update_stream_conf/2]). -export([readers/1]). --export([read_ahead_on/0]). +-export([read_ahead_on/0, read_ahead_limit/0]). -export([parse_offset_arg/1, filter_spec/1]). @@ -465,7 +465,8 @@ begin_stream(#stream_client{name = QName, Tag, Offset, Mode, AckRequired, Filter, Options0) when is_pid(LocalPid) -> CounterSpec = {{?MODULE, QName, Tag, self()}, []}, - Options1 = Options0#{read_ahead => read_ahead_on()}, + Options1 = Options0#{read_ahead => read_ahead_on(), + read_ahead_limit => read_ahead_limit()}, {ok, Seg0} = osiris:init_reader(LocalPid, Offset, CounterSpec, Options1), NextOffset = osiris_log:next_offset(Seg0) - 1, osiris:register_offset_listener(LocalPid, NextOffset), @@ -1522,3 +1523,21 @@ queue_vm_ets() -> read_ahead_on() -> application:get_env(rabbit, stream_read_ahead, true). + +-spec read_ahead_limit() -> integer() | undefined. +read_ahead_limit() -> + case application:get_env(rabbit, stream_read_ahead_limit, undefined) of + undefined -> + undefined; + Bytes when is_integer(Bytes) -> + Bytes; + Limit when is_list(Limit) -> + case rabbit_resource_monitor_misc:parse_information_unit(Limit) of + {ok, ParsedLimit} -> + ParsedLimit; + {error, parse_error} -> + ?LOG_ERROR("Unable to parse stream read ahead limit value " + "~tp", [Limit]), + undefined + end + end. diff --git a/deps/rabbit/test/config_schema_SUITE_data/rabbit.snippets b/deps/rabbit/test/config_schema_SUITE_data/rabbit.snippets index dc214a17076..5a4ffc6c5df 100644 --- a/deps/rabbit/test/config_schema_SUITE_data/rabbit.snippets +++ b/deps/rabbit/test/config_schema_SUITE_data/rabbit.snippets @@ -1253,6 +1253,26 @@ credential_validator.regexp = ^abc\\d+", [{rabbit, [ {stream_read_ahead, false} ]}], + []}, + + %% + %% Stream read limit + %% + {stream_read_ahead_limit_bytes, + " + stream.read_ahead_limit = 8192 + ", + [{rabbit, [ + {stream_read_ahead_limit, 8192} + ]}], + []}, + {stream_read_ahead_limit_information_unit, + " + stream.read_ahead_limit = 8KiB + ", + [{rabbit, [ + {stream_read_ahead_limit, "8KiB"} + ]}], []} ]. diff --git a/deps/rabbitmq_stream/src/rabbit_stream_reader.erl b/deps/rabbitmq_stream/src/rabbit_stream_reader.erl index aac8bd4abd8..575d587a6d3 100644 --- a/deps/rabbitmq_stream/src/rabbit_stream_reader.erl +++ b/deps/rabbitmq_stream/src/rabbit_stream_reader.erl @@ -2814,7 +2814,8 @@ init_reader(ConnectionTransport, CounterSpec = {{?MODULE, QueueResource, SubscriptionId, self()}, []}, Options0 = #{transport => ConnectionTransport, chunk_selector => get_chunk_selector(Properties), - read_ahead => rabbit_stream_queue:read_ahead_on()}, + read_ahead => rabbit_stream_queue:read_ahead_on(), + read_ahead_limit => rabbit_stream_queue:read_ahead_limit()}, Options1 = maps:merge(Options0, rabbit_stream_utils:filter_spec(Properties)), diff --git a/rabbitmq-components.mk b/rabbitmq-components.mk index 35ce5210dc2..61891fc7b63 100644 --- a/rabbitmq-components.mk +++ b/rabbitmq-components.mk @@ -49,7 +49,7 @@ dep_jose = hex 1.11.10 dep_khepri = hex 0.17.2 dep_khepri_mnesia_migration = hex 0.8.0 dep_meck = hex 1.0.0 -dep_osiris = git https://github.com/rabbitmq/osiris v1.10.0 +dep_osiris = git https://github.com/rabbitmq/osiris v1.10.1 dep_prometheus = hex 5.1.1 dep_ra = hex 2.17.1 dep_ranch = hex 2.2.0