Add stream.read_ahead_limit config option

(cherry picked from commit f40cecd)

# Conflicts:
#	deps/rabbit/src/rabbit_stream_queue.erl

Author: the-mikedavis

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]}
 ]}.

deps/rabbit/src/rabbit_stream_queue.erl

@@ -50,7 +50,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]).
@@ -468,7 +468,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),
@@ -1531,4 +1532,26 @@ shrink_all(_Node) ->
     {error, not_quorum_queue}.
 
 read_ahead_on() ->
-    application:get_env(rabbit, stream_read_ahead, true).
+<<<<<<< HEAD
+    application:get_env(rabbit, stream_read_ahead, true).
+=======
+    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.
+>>>>>>> f40cecd0a (Add `stream.read_ahead_limit` config option)

deps/rabbit/test/config_schema_SUITE_data/rabbit.snippets

@@ -1266,6 +1266,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"}
+     ]}],
    []}
 
 ].

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)),