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 fd3dbbd3 [SPARK-59926] Do not retry failed K8s event writes
fd3dbbd3 is described below
commit fd3dbbd304e3ba40405b25edfa892a584a47e7a7
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Thu Oct 1 11:00:18 2026 -0700
[SPARK-59926] Do not retry failed K8s event writes
### What changes were proposed in this pull request?
This PR aims to write K8s events with a Kubernetes client which does not
retry a failed request.
- `ConfigurableEventRecorder.withDefaultSink` uses
`KubernetesClientFactory.withoutRetries`, which is moved from
`KueueWorkloadUtils` for reuse. The derived client keeps the connections, the
interceptors and the request timeout of the operator client.
- `ReconcileError` is published on every retry, subject to
`spark.kubernetes.operator.events.minIntervalSeconds`, instead of only on the
first attempt, so that a dropped first write does not hide the whole failure.
- `docs/configuration.md` is updated accordingly.
### Why are the changes needed?
`DefaultEventSink` writes events synchronously on the reconciliation
thread. By default, the fabric8 client retries a write which fails with `429`,
`5xx` or an I/O error up to 10 times with backoff. So, a single event write can
hold a reconciliation for tens of seconds exactly when the API server is in
trouble and error events are published. Since a failed event write is only
logged anyway, there is no need to retry it.
### Does this PR introduce _any_ user-facing change?
No for the released versions, because K8s Events support is not released
yet in
[1.0.0](https://github.com/apache/spark-kubernetes-operator/releases/tag/1.0.0)
(2026-07-23).
### How was this patch tested?
Pass the CIs with the newly added and updated test cases.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #916 from dongjoon-hyun/SPARK-59926.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
docs/configuration.md | 15 +++-
.../operator/client/KubernetesClientFactory.java | 21 +++++
.../k8s/operator/kueue/KueueWorkloadUtils.java | 14 +---
.../operator/reconciler/SparkAppReconciler.java | 18 ++---
.../reconciler/SparkClusterReconciler.java | 18 ++---
.../operator/utils/ConfigurableEventRecorder.java | 16 +++-
.../spark/k8s/operator/utils/ReconcilerUtils.java | 14 ----
.../client/KubernetesClientFactoryTest.java | 82 +++++++++++++++++++
.../reconciler/SparkAppReconcilerTest.java | 9 +--
.../reconciler/SparkClusterReconcilerTest.java | 5 +-
.../ConfigurableEventRecorderDefaultSinkTest.java | 94 ++++++++++++++++++++++
11 files changed, 247 insertions(+), 59 deletions(-)
diff --git a/docs/configuration.md b/docs/configuration.md
index 526d491a..38b4ba69 100644
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -100,7 +100,7 @@ In addition, the operator publishes the following `Warning`
events.
| Reason | When |
|---|---|
-| `ReconcileError` | A reconciliation throws. Only the first attempt of a
failure episode publishes. |
+| `ReconcileError` | A reconciliation throws. It is republished on every retry
while the failure lasts, subject to [`minIntervalSeconds`](#event-frequency). |
| `CleanupError` | A cleanup throws, so the resource cannot finish deleting. |
| `StatusUpdateFailed` | A status patch is rejected. Transport-level errors
are skipped. |
| `KueueAdmissionRequestFailed` | Creating, reading or deleting a stale Kueue
`Workload` fails, or the driver or master cannot be read before the admission
is requested. Transport-level errors are skipped and retried every 5 seconds,
while a persistent failure is retried with the default interval. |
@@ -181,7 +181,13 @@ visible.
The interval is held from the moment an event is handed to the event sink, not
from the moment it
reaches the API server. The operator logs and swallows event write failures
rather than failing
-the reconciliation, so a write that failed still holds the interval; the next
repeat recovers it.
+the reconciliation, and its Kubernetes client does not retry an event write,
e.g. one that the API
+server throttles (`429`) or fails (`5xx`), since the reconciliation waits for
the write. So a
+write that failed is dropped, right away when the API server answers with an
error or refuses the
+connection, or after the request timeout (10 seconds by default) when it does
not answer, yet
+still holds the interval. An event which is republished, such as `SuspendHeld`
or a warning about
+a failure which lasts, recovers with its next repeat after the interval, while
one which is not,
+such as a state transition, is lost; the resource status remains the source of
truth.
## Metrics
@@ -223,6 +229,11 @@ via [Codahale JVM
Metrics](https://javadoc.io/doc/com.codahale.metrics/metrics-j
| kubernetes.client.`ResourceName`.`Method` | Meter |
Tracking the rates of HTTP request for a combination of one Kubernetes resource
and one http method |
| kubernetes.client.`NamespaceName`.`ResourceName`.`Method` | Meter |
Tracking the rates of HTTP request for a combination of one namespace-scoped
Kubernetes resource and one http method |
+A request which gets no response is counted in `kubernetes.client.failed`
while the client still
+has a retry left for it, even if it does not retry it, e.g. on a request
timeout. So a request of
+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.
+
### Latency for State Transition
Spark Operator also measures the latency between each state transition for
apps, in the format of
diff --git
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/client/KubernetesClientFactory.java
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/client/KubernetesClientFactory.java
index 9b975dff..8673ce72 100644
---
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/client/KubernetesClientFactory.java
+++
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/client/KubernetesClientFactory.java
@@ -23,6 +23,8 @@ import java.util.List;
import io.fabric8.kubernetes.client.Config;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.kubernetes.client.KubernetesClientBuilder;
+import io.fabric8.kubernetes.client.RequestConfig;
+import io.fabric8.kubernetes.client.RequestConfigBuilder;
import io.fabric8.kubernetes.client.http.HttpClient;
import io.fabric8.kubernetes.client.http.Interceptor;
import io.fabric8.kubernetes.client.utils.HttpClientUtils;
@@ -69,4 +71,23 @@ public final class KubernetesClientFactory {
})
.build();
}
+
+ /**
+ * Returns a client over the same connections, with the same interceptors
and request timeout,
+ * which does not retry a failed request. The given client retries a
throttled request, a server
+ * error and a connection failure with backoff, which holds the caller
meanwhile. Since fabric8
+ * calls {@link Interceptor#afterConnectionFailure} only while a request has
a retry left, the
+ * interceptors do not see a request of the returned client which gets no
response, so the client
+ * metrics do not count it as failed.
+ *
+ * @param client The client to derive from.
+ * @return A client which does not retry a failed request.
+ */
+ public static KubernetesClient withoutRetries(final KubernetesClient client)
{
+ RequestConfig requestConfig =
+ new RequestConfigBuilder(client.getConfiguration().getRequestConfig())
+ .withRequestRetryBackoffLimit(0)
+ .build();
+ return client.newClient(requestConfig).adapt(KubernetesClient.class);
+ }
}
diff --git
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadUtils.java
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadUtils.java
index 50099eb5..0e335a93 100644
---
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadUtils.java
+++
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadUtils.java
@@ -52,11 +52,10 @@ import io.fabric8.kubernetes.api.model.OwnerReference;
import io.fabric8.kubernetes.api.model.Toleration;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.kubernetes.client.KubernetesClientException;
-import io.fabric8.kubernetes.client.RequestConfig;
-import io.fabric8.kubernetes.client.RequestConfigBuilder;
import lombok.extern.slf4j.Slf4j;
import org.apache.spark.k8s.operator.Constants;
+import org.apache.spark.k8s.operator.client.KubernetesClientFactory;
import org.apache.spark.k8s.operator.context.BaseContext;
import org.apache.spark.k8s.operator.kueue.v1beta2.PodSet;
import org.apache.spark.k8s.operator.kueue.v1beta2.PodSetAssignment;
@@ -875,7 +874,7 @@ public final class KueueWorkloadUtils {
if (!shouldRecordPodsReady(context, cached)) {
return Optional.empty();
}
- KubernetesClient client = withoutRetries(context.getClient());
+ KubernetesClient client =
KubernetesClientFactory.withoutRetries(context.getClient());
String name = cached.getMetadata().getName();
try {
Workload workload = client.resource(cached).get();
@@ -909,15 +908,6 @@ public final class KueueWorkloadUtils {
}
}
- /** Returns a client over the same connections which does not retry a failed
request. */
- private static KubernetesClient withoutRetries(final KubernetesClient
client) {
- RequestConfig requestConfig =
- new RequestConfigBuilder(client.getConfiguration().getRequestConfig())
- .withRequestRetryBackoffLimit(0)
- .build();
- return client.newClient(requestConfig).adapt(KubernetesClient.class);
- }
-
/**
* Checks whether the given Workload is admitted without the `PodsReady`
condition, while as many
* pods of each of its pod sets as its count are ready.
diff --git
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconciler.java
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconciler.java
index 8cf56fcd..24ac7629 100644
---
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconciler.java
+++
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconciler.java
@@ -22,7 +22,6 @@ package org.apache.spark.k8s.operator.reconciler;
import static
org.apache.spark.k8s.operator.Constants.LABEL_SPARK_APPLICATION_NAME;
import static
org.apache.spark.k8s.operator.config.SparkOperatorConf.KUEUE_ENABLED;
import static
org.apache.spark.k8s.operator.reconciler.ReconcileProgress.completeAndDefaultRequeue;
-import static
org.apache.spark.k8s.operator.utils.ReconcilerUtils.isFirstAttempt;
import static
org.apache.spark.k8s.operator.utils.Utils.basicLabelSecondaryToPrimaryMapper;
import static
org.apache.spark.k8s.operator.utils.Utils.commonResourceLabelsStr;
@@ -141,15 +140,14 @@ public class SparkAppReconciler implements
Reconciler<SparkApplication>, Cleaner
retryInfo.isLastAttempt());
}
});
- // JOSDK calls this on every retry attempt. Emitting each time would
mean one blocking write
- // per attempt against an API server that is often the cause of the
failure, so only the
- // first attempt publishes: the reason is stable, so a later attempt
would only bump a count.
- if (isFirstAttempt(context)) {
- EventUtils.warn(
- context.eventRecorder(),
- EventUtils.REASON_RECONCILE_ERROR,
- "Spark App Reconciliation failed. " + EventUtils.describe(e));
- }
+ // JOSDK calls this on every retry attempt, and each one publishes. The
event write is not
+ // retried, so a write which a struggling API server dropped is
recovered by a later attempt,
+ // while ConfigurableEventRecorder drops a repeat with the same message
within the minimum
+ // interval, so a failure which persists only bumps the count of the one
event.
+ EventUtils.warn(
+ context.eventRecorder(),
+ EventUtils.REASON_RECONCILE_ERROR,
+ "Spark App Reconciliation failed. " + EventUtils.describe(e));
return ErrorStatusUpdateControl.noStatusUpdate();
} finally {
trackedMDC.reset();
diff --git
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkClusterReconciler.java
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkClusterReconciler.java
index 04b346f1..a35a9265 100644
---
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkClusterReconciler.java
+++
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkClusterReconciler.java
@@ -22,7 +22,6 @@ package org.apache.spark.k8s.operator.reconciler;
import static org.apache.spark.k8s.operator.Constants.LABEL_SPARK_CLUSTER_NAME;
import static
org.apache.spark.k8s.operator.config.SparkOperatorConf.KUEUE_ENABLED;
import static
org.apache.spark.k8s.operator.reconciler.ReconcileProgress.completeAndDefaultRequeue;
-import static
org.apache.spark.k8s.operator.utils.ReconcilerUtils.isFirstAttempt;
import static
org.apache.spark.k8s.operator.utils.Utils.basicLabelSecondaryToPrimaryMapper;
import java.util.ArrayList;
@@ -126,15 +125,14 @@ public class SparkClusterReconciler implements
Reconciler<SparkCluster>, Cleaner
retryInfo.isLastAttempt());
}
});
- // JOSDK calls this on every retry attempt. Emitting each time would
mean one blocking write
- // per attempt against an API server that is often the cause of the
failure, so only the
- // first attempt publishes: the reason is stable, so a later attempt
would only bump a count.
- if (isFirstAttempt(context)) {
- EventUtils.warn(
- context.eventRecorder(),
- EventUtils.REASON_RECONCILE_ERROR,
- "Spark Cluster Reconciliation failed. " + EventUtils.describe(e));
- }
+ // JOSDK calls this on every retry attempt, and each one publishes. The
event write is not
+ // retried, so a write which a struggling API server dropped is
recovered by a later attempt,
+ // while ConfigurableEventRecorder drops a repeat with the same message
within the minimum
+ // interval, so a failure which persists only bumps the count of the one
event.
+ EventUtils.warn(
+ context.eventRecorder(),
+ EventUtils.REASON_RECONCILE_ERROR,
+ "Spark Cluster Reconciliation failed. " + EventUtils.describe(e));
return ErrorStatusUpdateControl.noStatusUpdate();
} finally {
trackedMDC.reset();
diff --git
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/ConfigurableEventRecorder.java
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/ConfigurableEventRecorder.java
index c952e25f..7baff8a7 100644
---
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/ConfigurableEventRecorder.java
+++
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/ConfigurableEventRecorder.java
@@ -44,6 +44,8 @@ import
io.javaoperatorsdk.operator.api.event.ResourceEventRecorder;
import io.javaoperatorsdk.operator.api.reconciler.Context;
import lombok.extern.slf4j.Slf4j;
+import org.apache.spark.k8s.operator.client.KubernetesClientFactory;
+
/**
* An {@link EventRecorder} that drops every event unless {@link
*
org.apache.spark.k8s.operator.config.SparkOperatorConf#KUBERNETES_EVENTS_ENABLED}
is set, drops
@@ -102,13 +104,21 @@ public class ConfigurableEventRecorder implements
EventRecorder {
}
/**
- * Constructs a recorder over the JOSDK defaults, writing events with the
given client.
+ * Constructs a recorder over the JOSDK defaults, writing events with a
client derived from the
+ * given one which does not retry a failed request.
+ *
+ * <p>{@link DefaultEventSink} writes on the reconciliation thread, and an
error event often
+ * reports a failure of the API server itself, so a failed write is dropped
like any other write
+ * failure rather than retried with backoff while the reconciliation waits.
A write which the API
+ * server does not answer still waits for the request timeout of the given
client.
*
- * @param client The Kubernetes client used to write events.
+ * @param client The Kubernetes client to derive the client which writes
events from.
* @return A recorder gated on the events config option.
*/
public static ConfigurableEventRecorder withDefaultSink(KubernetesClient
client) {
- return new ConfigurableEventRecorder(new DefaultEventRecorder(new
DefaultEventSink(client)));
+ return new ConfigurableEventRecorder(
+ new DefaultEventRecorder(
+ new
DefaultEventSink(KubernetesClientFactory.withoutRetries(client))));
}
/**
diff --git
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/ReconcilerUtils.java
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/ReconcilerUtils.java
index 5388a646..60b32591 100644
---
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/ReconcilerUtils.java
+++
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/ReconcilerUtils.java
@@ -53,9 +53,7 @@ import io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.kubernetes.client.KubernetesClientException;
import io.fabric8.kubernetes.client.dsl.FilterWatchListDeletable;
import io.fabric8.kubernetes.client.dsl.PodResource;
-import io.javaoperatorsdk.operator.api.reconciler.Context;
import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
-import io.javaoperatorsdk.operator.api.reconciler.RetryInfo;
import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
import lombok.extern.slf4j.Slf4j;
@@ -357,18 +355,6 @@ public final class ReconcilerUtils {
};
}
- /**
- * Whether the given reconciliation is the first attempt at the resource
rather than a retry.
- * {@link RetryInfo#getAttemptCount()} is 0 for an execution that is not a
retry, and the info is
- * absent altogether when retries are not configured.
- *
- * @param context The reconciliation context.
- * @return True if this execution is not a retry, false otherwise.
- */
- public static boolean isFirstAttempt(Context<?> context) {
- return context.getRetryInfo().map(RetryInfo::getAttemptCount).orElse(0) ==
0;
- }
-
/**
* Whether the given failure is expected to clear without anyone acting on
it, so that a caller
* may wait it out rather than report it. A broken connection and an
overloaded or proxied server
diff --git
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/client/KubernetesClientFactoryTest.java
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/client/KubernetesClientFactoryTest.java
new file mode 100644
index 00000000..ba8dc4de
--- /dev/null
+++
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/client/KubernetesClientFactoryTest.java
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.k8s.operator.client;
+
+import static java.net.HttpURLConnection.HTTP_UNAVAILABLE;
+import static
org.apache.spark.k8s.operator.metrics.source.KubernetesMetricsInterceptor.HTTP_REQUEST_FAILED_GROUP;
+import static
org.apache.spark.k8s.operator.metrics.source.KubernetesMetricsInterceptor.HTTP_REQUEST_GROUP;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.util.List;
+
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.KubernetesClientException;
+import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
+import io.fabric8.kubernetes.client.server.mock.KubernetesMockServer;
+import org.junit.jupiter.api.Test;
+
+import
org.apache.spark.k8s.operator.metrics.source.KubernetesMetricsInterceptor;
+
+@EnableKubernetesMockClient
+class KubernetesClientFactoryTest {
+
+ private static final String CONFIG_MAP_PATH =
"/api/v1/namespaces/ns-1/configmaps/cm-1";
+
+ private KubernetesMockServer server;
+ private KubernetesClient client;
+
+ @Test
+ void withoutRetriesSendsAFailedRequestOnceThroughTheSameInterceptors() {
+ KubernetesMetricsInterceptor interceptor = new
KubernetesMetricsInterceptor();
+ try (KubernetesClient given =
+ KubernetesClientFactory.buildKubernetesClient(
+ List.of(interceptor), client.getConfiguration())) {
+ // The given client retries a request which fails once, and the retry
finds no ConfigMap.
+
server.expect().get().withPath(CONFIG_MAP_PATH).andReturn(HTTP_UNAVAILABLE,
null).once();
+
assertThat(given.configMaps().inNamespace("ns-1").withName("cm-1").get()).isNull();
+ assertThat(meterCount(interceptor, HTTP_REQUEST_GROUP)).isEqualTo(2L);
+
+
server.expect().get().withPath(CONFIG_MAP_PATH).andReturn(HTTP_UNAVAILABLE,
null).once();
+ KubernetesClient withoutRetries =
KubernetesClientFactory.withoutRetries(given);
+ assertThatThrownBy(
+ () ->
withoutRetries.configMaps().inNamespace("ns-1").withName("cm-1").get())
+ .isInstanceOfSatisfying(
+ KubernetesClientException.class,
+ e -> assertThat(e.getCode()).isEqualTo(HTTP_UNAVAILABLE));
+ // The single request is still measured by the interceptor of the given
client.
+ assertThat(meterCount(interceptor, HTTP_REQUEST_GROUP)).isEqualTo(3L);
+
+ // A connection failure is counted as a request but not as failed, since
fabric8 reports it
+ // to the interceptors only while the request has a retry left.
+ long failed = meterCount(interceptor, HTTP_REQUEST_FAILED_GROUP);
+ server.shutdown();
+ assertThatThrownBy(
+ () ->
withoutRetries.configMaps().inNamespace("ns-1").withName("cm-1").get())
+ .isInstanceOf(KubernetesClientException.class);
+ assertThat(meterCount(interceptor, HTTP_REQUEST_GROUP)).isEqualTo(4L);
+ assertThat(meterCount(interceptor,
HTTP_REQUEST_FAILED_GROUP)).isEqualTo(failed);
+ }
+ }
+
+ private static long meterCount(KubernetesMetricsInterceptor interceptor,
String name) {
+ return interceptor.metricRegistry().meter(name).getCount();
+ }
+}
diff --git
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconcilerTest.java
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconcilerTest.java
index d86f7b1e..572645dc 100644
---
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconcilerTest.java
+++
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconcilerTest.java
@@ -38,7 +38,6 @@ import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockConstruction;
-import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -422,7 +421,7 @@ class SparkAppReconcilerTest {
}
@Test
- void updateErrorStatusPublishesOnlyOnTheFirstAttempt() {
+ void updateErrorStatusPublishesOnARetryToo() {
when(mockContext.eventRecorder()).thenReturn(mockEventRecorder);
var retryInfo = mock(RetryInfo.class);
when(retryInfo.getAttemptCount()).thenReturn(1);
@@ -430,9 +429,9 @@ class SparkAppReconcilerTest {
reconciler.updateErrorStatus(app, mockContext, new
RuntimeException("boom"));
- // JOSDK calls this on every retry attempt, each emit is a blocking write
against an API
- // server that is often the cause of the failure.
- verify(mockEventRecorder, never()).record(any(EventRecord.class));
+ // The event write is not retried, so a retry attempt recovers a write
which a struggling API
+ // server dropped, while the recorder drops an unchanged repeat within the
minimum interval.
+
assertThat(captureRecordedEvent().reason()).isEqualTo(EventUtils.REASON_RECONCILE_ERROR);
}
@Test
diff --git
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/SparkClusterReconcilerTest.java
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/SparkClusterReconcilerTest.java
index 57e4a594..41ca91b2 100644
---
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/SparkClusterReconcilerTest.java
+++
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/SparkClusterReconcilerTest.java
@@ -33,7 +33,6 @@ import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockConstruction;
-import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -298,7 +297,7 @@ class SparkClusterReconcilerTest {
}
@Test
- void updateErrorStatusPublishesOnlyOnTheFirstAttempt() {
+ void updateErrorStatusPublishesOnARetryToo() {
when(mockContext.eventRecorder()).thenReturn(mockEventRecorder);
var retryInfo = mock(RetryInfo.class);
when(retryInfo.getAttemptCount()).thenReturn(1);
@@ -306,7 +305,7 @@ class SparkClusterReconcilerTest {
reconciler.updateErrorStatus(cluster, mockContext, new
RuntimeException("boom"));
- verify(mockEventRecorder, never()).record(any(EventRecord.class));
+
assertThat(captureRecordedEvent().reason()).isEqualTo(EventUtils.REASON_RECONCILE_ERROR);
}
@Test
diff --git
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/utils/ConfigurableEventRecorderDefaultSinkTest.java
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/utils/ConfigurableEventRecorderDefaultSinkTest.java
new file mode 100644
index 00000000..2fc6d1c0
--- /dev/null
+++
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/utils/ConfigurableEventRecorderDefaultSinkTest.java
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.k8s.operator.utils;
+
+import static java.net.HttpURLConnection.HTTP_UNAVAILABLE;
+import static org.apache.spark.k8s.operator.Constants.HTTP_TOO_MANY_REQUESTS;
+import static
org.apache.spark.k8s.operator.config.SparkOperatorConf.KUBERNETES_EVENTS_ENABLED;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+
+import java.util.Map;
+
+import io.fabric8.kubernetes.api.model.Event;
+import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
+import io.fabric8.kubernetes.client.server.mock.KubernetesMockServer;
+import io.javaoperatorsdk.operator.api.config.ConfigurationService;
+import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
+import io.javaoperatorsdk.operator.api.event.EventRecord;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import org.apache.spark.k8s.operator.SparkApplication;
+import org.apache.spark.k8s.operator.config.SparkOperatorConfManager;
+
+/**
+ * Tests {@link ConfigurableEventRecorder#withDefaultSink} against a mock API
server. It is
+ * separate from {@link ConfigurableEventRecorderTest}, since the mock server
extension starts a
+ * server for every test of the class which it annotates.
+ */
+@EnableKubernetesMockClient(crud = true)
+class ConfigurableEventRecorderDefaultSinkTest {
+
+ private KubernetesMockServer server;
+ private KubernetesClient client;
+
+ @AfterEach
+ void resetConf() {
+ SparkOperatorConfManager.INSTANCE.refresh(Map.of());
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = {HTTP_TOO_MANY_REQUESTS, HTTP_UNAVAILABLE})
+ void dropsAFailedWriteRatherThanRetryingIt(int code) {
+ // Only the first create fails, so a retry would have created the Event.
+
SparkOperatorConfManager.INSTANCE.refresh(Map.of(KUBERNETES_EVENTS_ENABLED.getKey(),
"true"));
+ ConfigurableEventRecorder recorder =
ConfigurableEventRecorder.withDefaultSink(client);
+ Context<?> context = appContext();
+
server.expect().post().withPath("/api/v1/namespaces/ns-1/events").andReturn(code,
null).once();
+
+ recorder.record(EventRecord.warning("ReconcileError", "dropped"), context);
+
assertThat(client.v1().events().inNamespace("ns-1").list().getItems()).isEmpty();
+
+ // The next write goes through, so the first one was dropped only for the
injected failure.
+ recorder.record(EventRecord.warning("ReconcileError", "published"),
context);
+ assertThat(client.v1().events().inNamespace("ns-1").list().getItems())
+ .extracting(Event::getMessage)
+ .containsExactly("published");
+ }
+
+ /** Returns a context from which the JOSDK recorder can build an Event about
app-1 in ns-1. */
+ private static Context<?> appContext() {
+ SparkApplication app = new SparkApplication();
+ app.setMetadata(
+ new
ObjectMetaBuilder().withName("app-1").withNamespace("ns-1").withUid("uid-1").build());
+ ControllerConfiguration<?> configuration =
mock(ControllerConfiguration.class);
+
doReturn(mock(ConfigurationService.class)).when(configuration).getConfigurationService();
+ Context<?> context = mock(Context.class);
+ doReturn(app).when(context).getPrimaryResource();
+ doReturn(configuration).when(context).getControllerConfiguration();
+ return context;
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
