Merge pull request #29989 from vbotbuildovich/backport-pr-29970-v26.1.x-827

[v26.1.x] k/fetch: Add fetch_response_dropped_bytes metric

Author: Lazin

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

@@ -460,7 +460,13 @@ static void fill_fetch_responses(
             resp_units = std::move(res.memory_units);
             resp.records = batch_reader(std::move(res).release_data());
         } else {
-            // TODO: add probe to measure how much of read data is discarded
+            if (res.has_data()) {
+                // Data was read from cloud storage but cannot fit in the
+                // response budget. This is pure read amplification: S3 bytes
+                // were downloaded, materialized, and now dropped.
+                octx.rctx.probe().add_fetch_response_dropped_bytes(
+                  res.data_size_bytes());
+            }
             resp.records = batch_reader();
         }
 

src/v/kafka/server/kafka_probe.h

@@ -113,6 +113,20 @@ class kafka_probe {
               "in the future or in the past"))},
           {},
           {sm::shard_label});
+
+        _metrics.add_group(
+          "kafka",
+          {sm::make_counter(
+            "fetch_response_dropped_bytes",
+            [this] { return _fetch_response_dropped_bytes; },
+            sm::description(
+              "Total bytes read from storage (including cloud storage) that "
+              "were discarded in fill_fetch_responses because the response "
+              "budget was already consumed by other partitions. Non-zero "
+              "values indicate read amplification: data was fetched from S3 "
+              "but not delivered to the consumer."))},
+          {},
+          {sm::shard_label});
     }
 
     void setup_public_metrics() {
@@ -153,6 +167,10 @@ class kafka_probe {
         _fetch_latency.record(micros.count());
     }
 
+    void add_fetch_response_dropped_bytes(uint64_t bytes) {
+        _fetch_response_dropped_bytes += bytes;
+    }
+
     void record_batch(uint64_t size, model::compression compression) {
         _batch_size.record(size);
         if (
@@ -167,6 +185,7 @@ class kafka_probe {
     friend prod_consume_fixture;
 
     uint32_t _produce_bad_create_time = 0;
+    uint64_t _fetch_response_dropped_bytes = 0;
 
     hist_t _produce_latency;
     hist_t _fetch_latency;