diff --git a/src/v/kafka/server/handlers/fetch.cc b/src/v/kafka/server/handlers/fetch.cc
index 3d6f428d5ad1d..9aecf7d9f1dff 100644
--- a/src/v/kafka/server/handlers/fetch.cc
+++ b/src/v/kafka/server/handlers/fetch.cc
@@ -35,8 +35,10 @@
 
 #include <seastar/core/do_with.hh>
 #include <seastar/core/future.hh>
+#include <seastar/core/scheduling.hh>
 #include <seastar/core/sleep.hh>
 #include <seastar/core/thread.hh>
+#include <seastar/core/with_scheduling_group.hh>
 #include <seastar/util/log.hh>
 
 #include <boost/range/irange.hpp>
@@ -555,22 +557,27 @@ fetch_handler::handle(request_context rctx, ss::smp_service_group ssg) {
     return ss::do_with(
       std::make_unique<op_context>(std::move(rctx), ssg),
       [](std::unique_ptr<op_context>& octx_ptr) {
-          auto& octx = *octx_ptr;
-          log_request(octx.rctx.header(), octx.request);
-          // top-level error is used for session-level errors
-          if (octx.session_ctx.has_error()) {
-              octx.response.data.error_code = octx.session_ctx.error();
-              return std::move(octx).send_response();
-          }
-          octx.response.data.error_code = error_code::none;
-          // first fetch, do not wait
-          return fetch_topic_partitions(octx)
-            .then([&octx] {
-                return ss::do_until(
-                  [&octx] { return octx.should_stop_fetch(); },
-                  [&octx] { return fetch_topic_partitions(octx); });
-            })
-            .then([&octx] { return std::move(octx).send_response(); });
+          auto sg
+            = octx_ptr->rctx.connection()->server().fetch_scheduling_group();
+          return ss::with_scheduling_group(sg, [&octx_ptr] {
+              auto& octx = *octx_ptr;
+
+              log_request(octx.rctx.header(), octx.request);
+              // top-level error is used for session-level errors
+              if (octx.session_ctx.has_error()) {
+                  octx.response.data.error_code = octx.session_ctx.error();
+                  return std::move(octx).send_response();
+              }
+              octx.response.data.error_code = error_code::none;
+              // first fetch, do not wait
+              return fetch_topic_partitions(octx)
+                .then([&octx] {
+                    return ss::do_until(
+                      [&octx] { return octx.should_stop_fetch(); },
+                      [&octx] { return fetch_topic_partitions(octx); });
+                })
+                .then([&octx] { return std::move(octx).send_response(); });
+          });
       });
 }
 
diff --git a/src/v/kafka/server/server.cc b/src/v/kafka/server/server.cc
index a343bca391ab1..a37da6568c2af 100644
--- a/src/v/kafka/server/server.cc
+++ b/src/v/kafka/server/server.cc
@@ -77,6 +77,7 @@ namespace kafka {
 server::server(
   ss::sharded<net::server_configuration>* cfg,
   ss::smp_service_group smp,
+  ss::scheduling_group fetch_sg,
   ss::sharded<cluster::metadata_cache>& meta,
   ss::sharded<cluster::topics_frontend>& tf,
   ss::sharded<cluster::config_frontend>& cf,
@@ -99,6 +100,7 @@ server::server(
   ssx::thread_worker& tw) noexcept
   : net::server(cfg, klog)
   , _smp_group(smp)
+  , _fetch_scheduling_group(fetch_sg)
   , _topics_frontend(tf)
   , _config_frontend(cf)
   , _feature_table(ft)
diff --git a/src/v/kafka/server/server.h b/src/v/kafka/server/server.h
index 934bb7fd786bf..d97bc0eca10db 100644
--- a/src/v/kafka/server/server.h
+++ b/src/v/kafka/server/server.h
@@ -28,6 +28,7 @@
 #include "utils/ema.h"
 
 #include <seastar/core/future.hh>
+#include <seastar/core/scheduling.hh>
 #include <seastar/core/sharded.hh>
 #include <seastar/core/smp.hh>
 
@@ -38,6 +39,7 @@ class server final : public net::server {
     server(
       ss::sharded<net::server_configuration>*,
       ss::smp_service_group,
+      ss::scheduling_group,
       ss::sharded<cluster::metadata_cache>&,
       ss::sharded<cluster::topics_frontend>&,
       ss::sharded<cluster::config_frontend>&,
@@ -71,6 +73,9 @@ class server final : public net::server {
     ss::future<> apply(ss::lw_shared_ptr<net::connection>) final;
 
     ss::smp_service_group smp_group() const { return _smp_group; }
+    ss::scheduling_group fetch_scheduling_group() const {
+        return _fetch_scheduling_group;
+    }
     cluster::topics_frontend& topics_frontend() {
         return _topics_frontend.local();
     }
@@ -148,6 +153,7 @@ class server final : public net::server {
 
 private:
     ss::smp_service_group _smp_group;
+    ss::scheduling_group _fetch_scheduling_group;
     ss::sharded<cluster::topics_frontend>& _topics_frontend;
     ss::sharded<cluster::config_frontend>& _config_frontend;
     ss::sharded<features::feature_table>& _feature_table;
diff --git a/src/v/redpanda/tests/fixture.h b/src/v/redpanda/tests/fixture.h
index 488628b92ef15..282457fae386c 100644
--- a/src/v/redpanda/tests/fixture.h
+++ b/src/v/redpanda/tests/fixture.h
@@ -115,6 +115,7 @@ class redpanda_thread_fixture {
         proto = std::make_unique<kafka::server>(
           &configs,
           app.smp_service_groups.kafka_smp_sg(),
+          app.sched_groups.fetch_sg(),
           app.metadata_cache,
           app.controller->get_topics_frontend(),
           app.controller->get_config_frontend(),