@@ -259,6 +259,11 @@ has a retry left for it, even if it does not retry it, e.g. on a request timeout
 a client which never retries, i.e. a Kubernetes event write or the recording of the Kueue
 `PodsReady` condition, is not counted there when it gets no response.
 
+`kubernetes.client.http.response.latency.nanos` ends when the response headers arrive, so it does
+not include reading the body. A request re-sent after a `401` is measured from the `401`, whose
+latency can include the token refresh. WebSocket upgrades, e.g. of a watch, are not recorded, and
+a rejected one can be counted more than once in `kubernetes.client.http.response`.
+
 ### Latency for State Transition
 
 Spark Operator also measures the latency between each state transition for apps, in the format of
 
@@ -90,6 +90,10 @@ affect building the project, CI, tests or examples.
   `0.5` and `0.999` quantiles of the `operator_sdk_*` histograms are the median and the 99.9th
   percentile instead of the mean and the 99th percentile. A histogram or timer of a custom metrics
   source has no `_sum` unless it is a `SummingHistogram` or a `SummingTimer` ([SPARK-59935](https://issues.apache.org/jira/browse/SPARK-59935)).
+- Since 1.1.0, `kubernetes.client.http.response.latency.nanos` records the time from sending each
+  HTTP request until its response headers arrive. 1.0 recorded nearly zero regardless of the
+  latency. It no longer records a WebSocket upgrade response, e.g. of a watch, so its count can be
+  lower than that of `kubernetes.client.http.response` ([SPARK-59928](https://issues.apache.org/jira/browse/SPARK-59928)).
 
 ### SparkApplication
 
 
@@ -29,6 +29,7 @@
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
 
 import com.codahale.metrics.Histogram;
 import com.codahale.metrics.Meter;
@@ -90,6 +91,25 @@ public KubernetesMetricsInterceptor() {
     }
   }
 
+  /**
+   * Called before a request is sent, to allow wrapping the consumer of the response body.
+   *
+   * <p>Wraps the consumer to carry the request start time, which {@link #after(HttpRequest,
+   * HttpResponse, AsyncBody.Consumer)} finds with {@link AsyncBody.Consumer#unwrap(Class)}. fabric8
+   * passes to it the outermost consumer after all interceptors have wrapped it, so an interceptor
+   * which wraps the consumer after this one must delegate {@code unwrap(...)}, or no latency is
+   * recorded.
+   *
+   * @param consumer the consumer of the response body
+   * @param request the request about to be sent
+   * @return the consumer carrying the request start time
+   */
+  @Override
+  public AsyncBody.Consumer<List<ByteBuffer>> consumer(
+      AsyncBody.Consumer<List<ByteBuffer>> consumer, HttpRequest request) {
+    return new TimedConsumer(consumer, new AtomicLong(System.nanoTime()));
+  }
+
   /**
    * Called before a request to allow for the manipulation of the request
    *
@@ -102,21 +122,30 @@ public void before(BasicBuilder builder, HttpRequest request, RequestTags tags)
   }
 
   /**
-   * Called after a non-WebSocket HTTP response is received. The body might or might not be already
-   * consumed.
+   * Called after an HTTP response is received. The body might or might not be already consumed.
    *
    * <p>Should be used to analyze response codes and headers, original response shouldn't be
    * altered.
    *
+   * <p>fabric8 also calls this for a WebSocket upgrade response with a {@code null} consumer, and
+   * again for an HTTP request re-sent after {@link #afterFailure(BasicBuilder, HttpResponse,
+   * RequestTags)}. The latency of a re-sent response is measured from the previous response.
+   *
    * @param request the original request sent to the server.
    * @param response the response received from the server.
+   * @param consumer the consumer of the response body, or {@code null} for a WebSocket upgrade.
    */
   @Override
   public void after(
       HttpRequest request,
       HttpResponse<?> response,
       AsyncBody.Consumer<List<ByteBuffer>> consumer) {
-    updateResponseMetrics(response, System.nanoTime());
+    TimedConsumer timedConsumer = consumer == null ? null : consumer.unwrap(TimedConsumer.class);
+    if (timedConsumer != null) {
+      long now = System.nanoTime();
+      responseLatency.update(now - timedConsumer.startTimeNanos().getAndSet(now));
+    }
+    updateResponseMetrics(response);
   }
 
   /**
@@ -139,7 +168,8 @@ public CompletableFuture<Boolean> afterFailure(
   /**
    * Called after a connection attempt fails.
    *
-   * <p>This method will be invoked on each failed connection attempt.
+   * <p>fabric8 invokes this method only for a failed connection attempt which has a retry left, not
+   * for the last attempt.
    *
    * @param request the HTTP request.
    * @param failure the Java exception that caused the failure.
@@ -183,11 +213,9 @@ private void updateRequestMetrics(HttpRequest request) {
         });
   }
 
-  private void updateResponseMetrics(HttpResponse response, long startTimeNanos) {
+  private void updateResponseMetrics(HttpResponse response) {
     Objects.requireNonNull(response);
-    final long latency = System.nanoTime() - startTimeNanos;
     responseRateMeter.mark();
-    responseLatency.update(latency);
     getMeterByResponseCode(response.code()).mark();
     if (KUBERNETES_CLIENT_METRICS_GROUP_BY_RESPONSE_CODE_GROUP_ENABLED.getValue()) {
       responseCodeGroupMeters.get(response.code() / 100 - 1).mark();
@@ -229,4 +257,23 @@ public Optional<Pair<String, String>> parseNamespaceScopedResource(String path)
       return Optional.empty();
     }
   }
+
+  /**
+   * Consumer of a response body which carries the start time of its request, restarted at each
+   * response so that a re-sent request is measured from the previous response.
+   */
+  private record TimedConsumer(
+      AsyncBody.Consumer<List<ByteBuffer>> delegate, AtomicLong startTimeNanos)
+      implements AsyncBody.Consumer<List<ByteBuffer>> {
+    @Override
+    public void consume(List<ByteBuffer> value, AsyncBody asyncBody) throws Exception {
+      delegate.consume(value, asyncBody);
+    }
+
+    @Override
+    public <U> U unwrap(Class<U> target) {
+      U self = AsyncBody.Consumer.super.unwrap(target);
+      return self != null ? self : delegate.unwrap(target);
+    }
+  }
 }
@@ -19,19 +19,27 @@
 
 package org.apache.spark.k8s.operator.metrics.source;
 
+import static java.net.HttpURLConnection.HTTP_UNAUTHORIZED;
+import static org.awaitility.Awaitility.await;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
+import java.nio.ByteBuffer;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+import com.codahale.metrics.Histogram;
 import com.codahale.metrics.Meter;
 import com.codahale.metrics.Metric;
 import io.fabric8.kubernetes.api.model.ConfigMap;
 import io.fabric8.kubernetes.api.model.ObjectMeta;
+import io.fabric8.kubernetes.client.Config;
+import io.fabric8.kubernetes.client.ConfigBuilder;
 import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.http.AsyncBody;
 import io.fabric8.kubernetes.client.http.Interceptor;
+import io.fabric8.kubernetes.client.informers.SharedIndexInformer;
 import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
 import io.fabric8.kubernetes.client.server.mock.KubernetesMockServer;
 import org.junit.jupiter.api.AfterEach;
@@ -50,6 +58,9 @@
 @TestMethodOrder(MethodOrderer.OrderAnnotation.class)
 class KubernetesMetricsInterceptorTest {
 
+  private static final String CONFIG_MAP_PATH =
+      "/api/v1/namespaces/spark-system/configmaps/spark-job-operator-configuration";
+
   private KubernetesMockServer mockServer;
   private KubernetesClient kubernetesClient;
 
@@ -125,6 +136,103 @@ void testWhenKubernetesServerNotWorking() {
     }
   }
 
+  @Test
+  @Order(3)
+  void testResponseLatency() {
+    KubernetesMetricsInterceptor metricsInterceptor = new KubernetesMetricsInterceptor();
+    List<Interceptor> interceptors = List.of(metricsInterceptor);
+    try (KubernetesClient client =
+        KubernetesClientFactory.buildKubernetesClient(
+            interceptors, kubernetesClient.getConfiguration())) {
+      ConfigMap configMap = createConfigMap();
+      mockServer
+          .expect()
+          .get()
+          .delay(50)
+          .withPath(CONFIG_MAP_PATH)
+          .andReturn(200, configMap)
+          .once();
+      client.resource(configMap).get();
+
+      Map<String, Metric> map = metricsInterceptor.metricRegistry().getMetrics();
+      Histogram latency = (Histogram) map.get("http.response.latency.nanos");
+      Assertions.assertEquals(1, latency.getCount());
+      Assertions.assertTrue(latency.getSnapshot().getMax() >= 50_000_000L);
+      Assertions.assertTrue(latency.getSnapshot().getMax() < 30_000_000_000L);
+    }
+  }
+
+  @Test
+  @Order(4)
+  void testResponseLatencyOfResentRequest() {
+    KubernetesMetricsInterceptor metricsInterceptor = new KubernetesMetricsInterceptor();
+    List<Interceptor> interceptors = List.of(metricsInterceptor);
+    Config config =
+        new ConfigBuilder(kubernetesClient.getConfiguration())
+            .withOauthTokenProvider(() -> "token")
+            .build();
+    try (KubernetesClient client =
+        KubernetesClientFactory.buildKubernetesClient(interceptors, config)) {
+      ConfigMap configMap = createConfigMap();
+      mockServer
+          .expect()
+          .get()
+          .delay(200)
+          .withPath(CONFIG_MAP_PATH)
+          .andReturn(HTTP_UNAUTHORIZED, null)
+          .once();
+      mockServer.expect().get().withPath(CONFIG_MAP_PATH).andReturn(200, configMap).once();
+      client.resource(configMap).get();
+
+      Assertions.assertEquals(2, meterCount(metricsInterceptor, "http.response"));
+      Assertions.assertEquals(1, meterCount(metricsInterceptor, "http.response.401"));
+      Assertions.assertEquals(0, meterCount(metricsInterceptor, "failed"));
+      Map<String, Metric> map = metricsInterceptor.metricRegistry().getMetrics();
+      Histogram latency = (Histogram) map.get("http.response.latency.nanos");
+      Assertions.assertEquals(2, latency.getCount());
+      Assertions.assertTrue(latency.getSnapshot().getMax() >= 200_000_000L);
+      Assertions.assertTrue(latency.getSnapshot().getMin() < 200_000_000L);
+    }
+  }
+
+  @Test
+  @Order(5)
+  void testResponseLatencyOfWebSocketUpgrade() {
+    KubernetesMetricsInterceptor metricsInterceptor = new KubernetesMetricsInterceptor();
+    List<Interceptor> interceptors = List.of(metricsInterceptor);
+    try (KubernetesClient client =
+            KubernetesClientFactory.buildKubernetesClient(
+                interceptors, kubernetesClient.getConfiguration());
+        SharedIndexInformer<ConfigMap> informer =
+            client.configMaps().inNamespace("spark-system").inform()) {
+      Map<String, Metric> map = metricsInterceptor.metricRegistry().getMetrics();
+      await()
+          .untilAsserted(
+              () -> {
+                Assertions.assertEquals(2, meterCount(metricsInterceptor, "http.response"));
+                Assertions.assertEquals(1, meterCount(metricsInterceptor, "http.response.101"));
+                Assertions.assertEquals(
+                    1, ((Histogram) map.get("http.response.latency.nanos")).getCount());
+              });
+      Assertions.assertTrue(informer.isWatching());
+    }
+  }
+
+  @Test
+  @Order(6)
+  void testConsumerUnwrapsDelegate() {
+    AsyncBody.Consumer<List<ByteBuffer>> delegate = (value, asyncBody) -> {};
+    AsyncBody.Consumer<List<ByteBuffer>> consumer =
+        new KubernetesMetricsInterceptor().consumer(delegate, null);
+    Assertions.assertSame(delegate, consumer.unwrap(delegate.getClass()));
+  }
+
+  private static long meterCount(KubernetesMetricsInterceptor interceptor, String name) {
+    Meter meter = interceptor.metricRegistry().getMeters().get(name);
+    Assertions.assertNotNull(meter, name);
+    return meter.getCount();
+  }
+
   private static SparkApplication createSparkApplication() {
     ObjectMeta meta = new ObjectMeta();
     meta.setName("sample-spark-application");