kafka: remove existing fetch latency measurement

src/v/kafka/server/handlers/fetch.cc

@@ -422,8 +422,7 @@ read_result::memory_units_t reserve_memory_units(
 static void fill_fetch_responses(
   op_context& octx,
   std::vector<read_result> results,
-  const std::vector<op_context::response_placeholder_ptr>& responses,
-  op_context::latency_point start_time) {
+  const std::vector<op_context::response_placeholder_ptr>& responses) {
     auto range = boost::irange<size_t>(0, results.size());
     if (unlikely(results.size() != responses.size())) {
         // soft assert & recovery attempt
@@ -519,10 +518,6 @@ static void fill_fetch_responses(
         }
 
         resp_it->set(std::move(resp));
-        std::chrono::microseconds fetch_latency
-          = std::chrono::duration_cast<std::chrono::microseconds>(
-            op_context::latency_clock::now() - start_time);
-        octx.rctx.probe().record_fetch_latency(fetch_latency);
     }
 }
 
@@ -1075,10 +1070,7 @@ class nonpolling_fetch_plan_executor final : public fetch_plan_executor::impl {
             });
 
         fill_fetch_responses(
-          octx,
-          std::move(results.read_results),
-          fetch.responses,
-          fetch.start_time);
+          octx, std::move(results.read_results), fetch.responses);
 
         _last_result_size[fetch.shard] = results.total_size;
         _completed_shard_fetches.push_back(std::move(fetch));

src/v/kafka/server/handlers/fetch.h

@@ -331,9 +331,6 @@ struct read_result {
 // struct aggregating fetch requests and corresponding response iterators for
 // the same shard
 struct shard_fetch {
-    explicit shard_fetch(op_context::latency_point start_time)
-      : start_time{start_time} {}
-
     void push_back(
       ntp_fetch_config config, op_context::response_placeholder_ptr r_ph) {
         requests.push_back(std::move(config));
@@ -349,7 +346,6 @@ struct shard_fetch {
     ss::shard_id shard;
     std::vector<ntp_fetch_config> requests;
     std::vector<op_context::response_placeholder_ptr> responses;
-    op_context::latency_point start_time;
 
     friend std::ostream& operator<<(std::ostream& o, const shard_fetch& sf) {
         fmt::print(o, "{}", sf.requests);
@@ -358,10 +354,8 @@ struct shard_fetch {
 };
 
 struct fetch_plan {
-    explicit fetch_plan(
-      size_t shards,
-      op_context::latency_point start_time = op_context::latency_clock::now())
-      : fetches_per_shard(shards, shard_fetch(start_time)) {
+    explicit fetch_plan(size_t shards)
+      : fetches_per_shard(shards, shard_fetch()) {
         for (size_t i = 0; i < fetches_per_shard.size(); i++) {
             fetches_per_shard[i].shard = i;
         }