This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/spark-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new 30f2af3e [SPARK-60020] Fix `KubernetesMetricsInterceptor` exceptions
for `namespaces` paths and out-of-range response codes
30f2af3e is described below
commit 30f2af3e33a23f894ec6a1b261cb94a0fdb25f4c
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Wed Oct 7 08:45:40 2026 -0700
[SPARK-60020] Fix `KubernetesMetricsInterceptor` exceptions for
`namespaces` paths and out-of-range response codes
### What changes were proposed in this pull request?
This PR aims to fix two robustness bugs of `KubernetesMetricsInterceptor`.
- `parseNamespaceScopedResource` returns `Optional.empty()` unless the
request path has `namespaces/<namespace>/<resource>` segments. It matches the
`namespaces` path segment instead of the first `namespaces` substring.
- `updateResponseMetrics` marks a response code group meter, `1xx` to
`5xx`, only when the response code belongs to one of them.
### Why are the changes needed?
`parseNamespaceScopedResource` assumes that `<namespace>/<resource>` always
follows the first `namespaces` substring of the path. It throws for the
following paths.
| Path | Request | Exception |
| --- | --- | --- |
| `/api/v1/namespaces` | List or watch of namespaces |
`StringIndexOutOfBoundsException` |
| `/api/v1/namespaces/spark-system` | Get of a namespace |
`ArrayIndexOutOfBoundsException` |
| `/apis/rbac.authorization.k8s.io/v1/clusterroles/namespaces-reader` | Get
of a cluster-scoped resource whose name contains `namespaces` |
`ArrayIndexOutOfBoundsException` |
Since fabric8 7.8.0 `StandardHttpClient` calls `Interceptor.before(...)`
synchronously, the exception fails the request itself before it is sent, e.g.
`client.namespaces().list()` throws `StringIndexOutOfBoundsException`.
On `main`, the Kueue integration sends the last kind of request. It gets
the following cluster-scoped objects by name, and the client metrics are
enabled by default.
- `WorkloadPriorityClass` of the `kueue.x-k8s.io/priority-class` label
- `PriorityClass` of a pod template
- `ResourceFlavor` which Kueue assigned
For example, a `ResourceFlavor` named `all-namespaces` gives
`/apis/kueue.x-k8s.io/v1beta2/resourceflavors/all-namespaces`, which throws
`StringIndexOutOfBoundsException` like `/api/v1/namespaces`. Since the Kueue
code catches only `KubernetesClientException`, `IllegalStateException` and
`IllegalArgumentException`, it reaches the generic `catch (Exception e)` of
`AppInitStep` and `ClusterInitStep`, and the `SparkApplication` or
`SparkCluster` fails with `SchedulingFailure`. 1.0.0 has [...]
`updateResponseMetrics` uses `response.code() / 100 - 1` as the index of
the group meters. Netty accepts any non-negative status code, e.g. `HTTP/1.1
999` from a proxy, and fabric8 passes it to `Interceptor.after(...)`. For a
code outside of `100` to `599`, `responseCodeGroupMeters.get(...)` throws
`IndexOutOfBoundsException` on the Vert.x event loop. fabric8 calls
`after(...)` of all interceptors in one loop of a `thenAccept` callback whose
future it drops, so the exception silently [...]
Both bugs exist in the latest release.
- https://github.com/apache/spark-kubernetes-operator/releases/tag/1.0.0
(2026-07-26)
### Does this PR introduce _any_ user-facing change?
No for 1.0.0. On `main`, a `SparkApplication` or `SparkCluster` with Kueue
no longer fails with `SchedulingFailure` when the name of its priority class or
`ResourceFlavor` contains `namespaces`. Otherwise, the metrics are unchanged.
For a response code outside of `100` to `599`,
`kubernetes.client.http.response` and `kubernetes.client.http.response.<code>`
are recorded as before, without the exception.
### How was this patch tested?
Pass the CIs with the newly added test cases. Without this fix,
`testParseNamespaceScopedResource` fails with `StringIndexOutOfBoundsException`
and `testResponseCodeGroups` fails with `IndexOutOfBoundsException`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #947 from dongjoon-hyun/SPARK-60020.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../source/KubernetesMetricsInterceptor.java | 18 +++++----
.../source/KubernetesMetricsInterceptorTest.java | 46 ++++++++++++++++++++++
2 files changed, 57 insertions(+), 7 deletions(-)
diff --git
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java
index dff97e92..e159bd34 100644
---
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java
+++
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java
@@ -23,6 +23,7 @@ import static
org.apache.spark.k8s.operator.config.SparkOperatorConf.KUBERNETES_
import java.nio.ByteBuffer;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -190,7 +191,10 @@ public class KubernetesMetricsInterceptor implements
Interceptor, Source {
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();
+ int index = response.code() / 100 - 1;
+ if (index >= 0 && index < responseCodeGroupMeters.size()) {
+ responseCodeGroupMeters.get(index).mark();
+ }
}
}
@@ -217,14 +221,14 @@ public class KubernetesMetricsInterceptor implements
Interceptor, Source {
* Parses the given path to extract namespace-scoped resource information.
*
* @param path The request path.
- * @return An Optional containing a Pair of namespace and resource name, or
empty if not found.
+ * @return An Optional containing a Pair of namespace and resource name, or
empty if the path has
+ * no {@code namespaces/<namespace>/<resource>} segments.
*/
public Optional<Pair<String, String>> parseNamespaceScopedResource(String
path) {
- if (path.contains(NAMESPACES)) {
- int index = path.indexOf(NAMESPACES) + NAMESPACES.length();
- String namespaceAndResources = path.substring(index + 1);
- String[] parts = namespaceAndResources.split("/");
- return Optional.of(Pair.of(parts[0], parts[1]));
+ List<String> segments = Arrays.asList(path.split("/"));
+ int index = segments.indexOf(NAMESPACES);
+ if (index >= 0 && index + 2 < segments.size()) {
+ return Optional.of(Pair.of(segments.get(index + 1), segments.get(index +
2)));
} else {
return Optional.empty();
}
diff --git
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptorTest.java
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptorTest.java
index 36e16018..9fab0e18 100644
---
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptorTest.java
+++
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptorTest.java
@@ -20,17 +20,21 @@
package org.apache.spark.k8s.operator.metrics.source;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
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.KubernetesClient;
+import io.fabric8.kubernetes.client.http.HttpResponse;
import io.fabric8.kubernetes.client.http.Interceptor;
import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
import io.fabric8.kubernetes.client.server.mock.KubernetesMockServer;
@@ -45,6 +49,7 @@ import org.apache.spark.k8s.operator.SparkApplication;
import org.apache.spark.k8s.operator.client.KubernetesClientFactory;
import org.apache.spark.k8s.operator.metrics.SummingHistogram;
import org.apache.spark.k8s.operator.spec.ApplicationSpec;
+import org.apache.spark.util.Pair;
@EnableKubernetesMockClient(crud = true)
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
@@ -125,6 +130,47 @@ class KubernetesMetricsInterceptorTest {
}
}
+ @Test
+ @Order(3)
+ void testParseNamespaceScopedResource() {
+ KubernetesMetricsInterceptor metricsInterceptor = new
KubernetesMetricsInterceptor();
+ Assertions.assertEquals(
+ Optional.of(Pair.of("spark-system", "configmaps")),
+ metricsInterceptor.parseNamespaceScopedResource(
+
"/api/v1/namespaces/spark-system/configmaps/spark-job-operator-configuration"));
+ Assertions.assertEquals(
+ Optional.of(Pair.of("spark-test", "sparkapplications")),
+ metricsInterceptor.parseNamespaceScopedResource(
+
"/apis/spark.apache.org/v1/namespaces/spark-test/sparkapplications"));
+ for (String path :
+ List.of(
+ "/api/v1/namespaces",
+ "/api/v1/namespaces/spark-system",
+
"/apis/rbac.authorization.k8s.io/v1/clusterroles/namespaces-reader")) {
+ Assertions.assertEquals(
+ Optional.empty(),
metricsInterceptor.parseNamespaceScopedResource(path), path);
+ }
+ }
+
+ @Test
+ @Order(4)
+ void testResponseCodeGroups() {
+ KubernetesMetricsInterceptor metricsInterceptor = new
KubernetesMetricsInterceptor();
+ for (int code : new int[] {99, 100, 599, 600, 999}) {
+ HttpResponse<?> response = mock(HttpResponse.class);
+ when(response.code()).thenReturn(code);
+ metricsInterceptor.after(null, response, null);
+ }
+ Map<String, Meter> meters =
metricsInterceptor.metricRegistry().getMeters();
+ Assertions.assertEquals(5, meters.get("http.response").getCount());
+ Assertions.assertEquals(1, meters.get("http.response.999").getCount());
+ Assertions.assertEquals(1, meters.get("1xx").getCount());
+ Assertions.assertEquals(0, meters.get("2xx").getCount());
+ Assertions.assertEquals(0, meters.get("3xx").getCount());
+ Assertions.assertEquals(0, meters.get("4xx").getCount());
+ Assertions.assertEquals(1, meters.get("5xx").getCount());
+ }
+
private static SparkApplication createSparkApplication() {
ObjectMeta meta = new ObjectMeta();
meta.setName("sample-spark-application");
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
