Merge pull request redpanda-data#15328 from ballard26/offset-table

src/v/config/configuration.cc

@@ -587,6 +587,19 @@ configuration::configuration()
       "wasn't reached",
       {.needs_restart = needs_restart::no, .visibility = visibility::tunable},
       1ms)
+  , fetch_read_strategy(
+      *this,
+      "fetch_read_strategy",
+      "The strategy used to fulfill fetch requests",
+      {.needs_restart = needs_restart::no,
+       .example = model::fetch_read_strategy_to_string(
+         model::fetch_read_strategy::non_polling),
+       .visibility = visibility::tunable},
+      model::fetch_read_strategy::non_polling,
+      {
+        model::fetch_read_strategy::polling,
+        model::fetch_read_strategy::non_polling,
+      })
   , alter_topic_cfg_timeout_ms(
       *this,
       "alter_topic_cfg_timeout_ms",

src/v/config/configuration.h

@@ -139,6 +139,7 @@ struct configuration final : public config_store {
     property<std::chrono::milliseconds> tx_timeout_delay_ms;
     deprecated_property rm_violation_recovery_policy;
     property<std::chrono::milliseconds> fetch_reads_debounce_timeout;
+    enum_property<model::fetch_read_strategy> fetch_read_strategy;
     property<std::chrono::milliseconds> alter_topic_cfg_timeout_ms;
     property<model::cleanup_policy_bitflags> log_cleanup_policy;
     enum_property<model::timestamp_type> log_message_timestamp_type;

src/v/config/convert.h

@@ -500,4 +500,35 @@ struct convert<pandaproxy::schema_registry::schema_id_validation_mode> {
     }
 };
 
+template<>
+struct convert<model::fetch_read_strategy> {
+    using type = model::fetch_read_strategy;
+
+    static constexpr auto acceptable_values = std::to_array(
+      {model::fetch_read_strategy_to_string(type::polling),
+       model::fetch_read_strategy_to_string(type::non_polling)});
+
+    static Node encode(const type& rhs) { return Node(fmt::format("{}", rhs)); }
+
+    static bool decode(const Node& node, type& rhs) {
+        auto value = node.as<std::string>();
+
+        if (
+          std::find(acceptable_values.begin(), acceptable_values.end(), value)
+          == acceptable_values.end()) {
+            return false;
+        }
+
+        rhs = string_switch<type>(std::string_view{value})
+                .match(
+                  model::fetch_read_strategy_to_string(type::polling),
+                  type::polling)
+                .match(
+                  model::fetch_read_strategy_to_string(type::non_polling),
+                  type::non_polling);
+
+        return true;
+    }
+};
+
 } // namespace YAML

src/v/config/property.h

@@ -640,6 +640,8 @@ consteval std::string_view property_type_name() {
                            pandaproxy::schema_registry::
                              schema_id_validation_mode>) {
         return "string";
+    } else if constexpr (std::is_same_v<type, model::fetch_read_strategy>) {
+        return "string";
     } else {
         static_assert(
           utils::unsupported_type<T>::value, "Type name not defined");

src/v/config/rjson_serialization.cc

@@ -160,6 +160,11 @@ void rjson_serialize(
     stringize(w, v);
 }
 
+void rjson_serialize(
+  json::Writer<json::StringBuffer>& w, const model::fetch_read_strategy& v) {
+    stringize(w, v);
+}
+
 void rjson_serialize(
   json::Writer<json::StringBuffer>& w,
   const model::cloud_storage_chunk_eviction_strategy& v) {

src/v/config/rjson_serialization.h

@@ -76,6 +76,9 @@ void rjson_serialize(
 void rjson_serialize(
   json::Writer<json::StringBuffer>& w, const model::leader_balancer_mode& v);
 
+void rjson_serialize(
+  json::Writer<json::StringBuffer>& w, const model::fetch_read_strategy& v);
+
 void rjson_serialize(
   json::Writer<json::StringBuffer>& w,
   const model::cloud_storage_chunk_eviction_strategy& v);