[SPARK-53664] Update updateResponseMetrics to handle valid responses only

### What changes were proposed in this pull request?

This PR aims to make `updateResponseMetrics` to handle valid responses only.

### Why are the changes needed?

To simplify the method logic.

### Does this PR introduce _any_ user-facing change?

No behavior change.

### How was this patch tested?

Pass the CIs.

### Was this patch authored or co-authored using generative AI tooling?

No.

Closes apache#331 from dongjoon-hyun/SPARK-53664.

Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>

Author: dongjoon-hyun

spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java

@@ -25,6 +25,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
@@ -129,7 +130,7 @@ public void after(
   @Override
   public CompletableFuture<Boolean> afterFailure(
       BasicBuilder builder, HttpResponse<?> response, RequestTags tags) {
-    updateResponseMetrics(null, System.nanoTime());
+    requestFailedRateMeter.mark();
     return CompletableFuture.completedFuture(false);
   }
 
@@ -143,7 +144,7 @@ public CompletableFuture<Boolean> afterFailure(
    */
   @Override
   public void afterConnectionFailure(HttpRequest request, Throwable failure) {
-    updateResponseMetrics(null, System.nanoTime());
+    requestFailedRateMeter.mark();
   }
 
   /**
@@ -181,16 +182,13 @@ private void updateRequestMetrics(HttpRequest request) {
   }
 
   private void updateResponseMetrics(HttpResponse response, long startTimeNanos) {
+    Objects.requireNonNull(response);
     final long latency = System.nanoTime() - startTimeNanos;
-    if (response != null) {
-      this.responseRateMeter.mark();
-      this.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();
-      }
-    } else {
-      this.requestFailedRateMeter.mark();
+    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();
     }
   }