Revert ct read debounce

src/v/cloud_topics/BUILD

@@ -133,7 +133,6 @@ redpanda_cc_library(
         "//src/v/cloud_topics/level_zero/cluster_services_impl",
         "//src/v/cloud_topics/level_zero/pipeline:read_pipeline",
         "//src/v/cloud_topics/level_zero/pipeline:write_pipeline",
-        "//src/v/cloud_topics/level_zero/read_debounce",
         "//src/v/cloud_topics/level_zero/read_fanout",
         "//src/v/cloud_topics/level_zero/read_request_scheduler",
         "//src/v/cloud_topics/level_zero/reader:fetch_handler",

src/v/cloud_topics/data_plane_impl.cc

@@ -19,7 +19,6 @@
 #include "cloud_topics/level_zero/cluster_services_impl/cluster_services.h"
 #include "cloud_topics/level_zero/pipeline/read_pipeline.h"
 #include "cloud_topics/level_zero/pipeline/write_pipeline.h"
-#include "cloud_topics/level_zero/read_debounce/read_debounce.h"
 #include "cloud_topics/level_zero/read_fanout/read_fanout.h"
 #include "cloud_topics/level_zero/read_request_scheduler/read_request_scheduler.h"
 #include "cloud_topics/level_zero/reader/fetch_request_handler.h"
@@ -82,12 +81,6 @@ class impl
                   return _read_pipeline.local().register_read_pipeline_stage();
               }));
         }
-        if (config::shard_local_cfg().cloud_topics_fetch_debounce_enabled()) {
-            co_await construct_service(
-              _read_debounce, ss::sharded_parameter([this] {
-                  return _read_pipeline.local().register_read_pipeline_stage();
-              }));
-        }
 
         co_await construct_service(
           _fetch_handler,
@@ -113,10 +106,6 @@ class impl
             co_await _read_request_scheduler.invoke_on_all(
               [](auto& s) { return s.start(); });
         }
-        if (_read_debounce.local_is_initialized()) {
-            co_await _read_debounce.invoke_on_all(
-              [](auto& s) { return s.start(); });
-        }
         co_await _fetch_handler.invoke_on_all(
           [](auto& s) { return s.start(); });
         co_await _batch_cache.invoke_on_all([](auto& s) { return s.start(); });
@@ -185,7 +174,6 @@ class impl
     ss::sharded<l0::read_pipeline<>> _read_pipeline;
     ss::sharded<l0::read_fanout> _read_fanout;
     ss::sharded<l0::read_request_scheduler> _read_request_scheduler;
-    ss::sharded<l0::read_debounce<>> _read_debounce;
 
     ss::sharded<l0::fetch_handler> _fetch_handler;
     // Batch cache

src/v/cloud_topics/level_zero/pipeline/BUILD

@@ -5,8 +5,6 @@ package(default_visibility = [
     "//src/v/cloud_topics:__pkg__",
     "//src/v/cloud_topics/level_zero/batcher:__pkg__",
     "//src/v/cloud_topics/level_zero/batcher/tests:__pkg__",
-    "//src/v/cloud_topics/level_zero/read_debounce:__pkg__",
-    "//src/v/cloud_topics/level_zero/read_debounce/tests:__pkg__",
     "//src/v/cloud_topics/level_zero/read_fanout:__pkg__",
     "//src/v/cloud_topics/level_zero/read_fanout/tests:__pkg__",
     "//src/v/cloud_topics/level_zero/read_request_scheduler:__pkg__",