Skip to content

Commit 89f97fd

Browse files
committed
[SPARK-59694] Do not treat an unanswered request as a permanent failure
### What changes were proposed in this pull request? This PR makes `ReconcilerUtils.isTransientError` classify the failures fabric8 reports without a response status, and makes `shouldBackoffBeforeRetry` delay before retrying them. `KubernetesClientException.getCode()` returns `-1`, not `0`, when no response status was received, so the existing `case 0` never matched anything the client produces. This bug was introduced by the following. - #530 https://github.com/apache/spark-kubernetes-operator/blob/957d4bc2919abe7c863cace3f4dd773f0ade5fe3/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/ReconcilerUtils.java#L306-L308 A new `NO_RESPONSE_CODE` constant carries that value, and a new private `brokeOnTheWayToTheApiServer` decides whether such a failure left the request unanswered. It is written as the complement of the failures that repeating cannot change — an unparsable answer, a handshake rejected over a certificate, and a rejection raised before sending, which carries no cause at all — so it does not depend on the HTTP client in use. That matters because the pinned client reports a connection the peer closes mid-response with a type of its own that is not even an `IOException`. `shouldBackoffBeforeRetry` gains the same code: an unanswered request carries no `Status` and so can never supply the `Retry-After` hint that is the only other way to earn a delay. ### Why are the changes needed? No failure ever reported code `0`, so every failure that did not reach a healthy API server was treated as a decision by that API server. Verified against the pinned client by pointing a real `KubernetesClient` at local sockets: | Failure | `getCode()` | Innermost cause | Classified | | --- | --- | --- | --- | | Connection refused | `-1` | `java.net.ConnectException` | transient | | Connect timeout | `-1` | `io.netty.channel.ConnectTimeoutException` | transient | | Read timeout | `-1` | `io.vertx.core.impl.NoStackTraceTimeoutException` | transient | | Connection reset | `-1` | `java.net.SocketException` | transient | | Peer closed mid-response | `-1` | `io.vertx.core.http.HttpClosedException` | transient | | DNS failure | `-1` | `java.net.UnknownHostException` | transient | | Unparsable response body | `-1` | `com.fasterxml.jackson.databind.exc.MismatchedInputException` | permanent | | Rejected before sending | `-1` | none | permanent | The misclassification reached every caller: | Caller | Before | After | | --- | --- | --- | | `getOrCreateSecondaryResource` | rethrew on the first attempt, skipping the `GET` that avoids a duplicate create, which `AppInitStep` turns into a terminal `SchedulingFailure` under the default `RestartPolicy.Never` | retries with backoff | | `getResourceStrictly` | reported an unanswered read as a refusal | reports it as one the create path settles | | `StatusRecorder.persistStatus` | wrote a `StatusUpdateFailed` event per blip, adding load to the server that caused it | stays quiet | | `KueueWorkloadUtils.retryAfterRequestFailure` | warning event plus the default requeue interval | the short interval, no event | The classification also holds in the other direction: an unparsable response and a rejection raised before sending no longer pass as transient, so they keep their warning event and longer requeue instead of being retried forever in silence. ### Does this PR introduce _any_ user-facing change? Yes. A `SparkApplication` that loses its connection to the API server while requesting driver resources is now retried with backoff instead of failing on the first attempt, and such failures no longer emit `Warning` events. Retries stay bounded by `spark.kubernetes.operator.api.secondaryResourceCreateMaxAttempts` (3 by default), so exhausting them still ends in `SchedulingFailure`. `SparkCluster` master and worker resources are created through `serverSideApply()` rather than `getOrCreateSecondaryResource()`, so they are unaffected; a `SparkCluster` benefits through its status updates and its Kueue admission requests. ### How was this patch tested? Pass the CIs. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Opus 5 Closes #860 from dongjoon-hyun/SPARK-59694. Authored-by: Dongjoon Hyun <dongjoon@apache.org> Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>
1 parent 957d4bc commit 89f97fd

6 files changed

Lines changed: 372 additions & 26 deletions

File tree

‎docs/config_properties.md‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -6,8 +6,8 @@
66
| spark.kubernetes.operator.api.retryMaxAttempts | Integer | 15 | false | Max attempts of retries on unhandled controller errors. Setting this to non-positive value means no retry. |
77
| spark.kubernetes.operator.api.secondaryResourceCreateBackoffJitterMillis | Long | 500 | false | Jitter (in milliseconds) added to backoff delay between retries when creating secondary resources for Spark application. |
88
| spark.kubernetes.operator.api.secondaryResourceCreateBackoffMultiplier | Double | 2.0 | false | Backoff multiplier applied to each retry interval when creating secondary resources for Spark application. |
9-
| spark.kubernetes.operator.api.secondaryResourceCreateInitialBackoffMillis | Long | 1000 | false | Initial backoff (in milliseconds) between retries when creating secondary resources for Spark application. Backoff is applied whenever the API server response carries a Retry-After hint, or on 409 (Conflict) and 429 (Too Many Requests) responses. |
10-
| spark.kubernetes.operator.api.secondaryResourceCreateMaxAttempts | Long | 3 | false | Maximal number of retry attempts of requesting secondary resource for Spark application. This would be performed on top of k8s client spark.kubernetes.operator.retry.maxAttempts to overcome potential conflicting reconcile on the same SparkApplication, as well as API server errors (408/500/502/503/504) and network-level timeouts. Exponential backoff with jitter is applied before retrying whenever the API server response carries a Retry-After hint, or on 409 (Conflict) and 429 (Too Many Requests) responses. Other retryable errors without a Retry-After hint are retried immediately. This should be positive number. |
9+
| spark.kubernetes.operator.api.secondaryResourceCreateInitialBackoffMillis | Long | 1000 | false | Initial backoff (in milliseconds) between retries when creating secondary resources for Spark application. Backoff is applied whenever the API server response carries a Retry-After hint, on 409 (Conflict) and 429 (Too Many Requests) responses, and on a request the API server never answered, which carries no response status and so can never supply a Retry-After hint. |
10+
| spark.kubernetes.operator.api.secondaryResourceCreateMaxAttempts | Long | 3 | false | Maximal number of retry attempts of requesting secondary resource for Spark application. This would be performed on top of k8s client spark.kubernetes.operator.retry.maxAttempts to overcome potential conflicting reconcile on the same SparkApplication, as well as API server errors (408/500/502/503/504) and requests the API server never answered. Exponential backoff with jitter is applied before retrying whenever the API server response carries a Retry-After hint, on 409 (Conflict) and 429 (Too Many Requests) responses, and on a request that went unanswered, which carries no response status and so can never supply a Retry-After hint. The remaining retryable errors are retried immediately. This should be positive number. |
1111
| spark.kubernetes.operator.api.secondaryResourceCreateMaxBackoffMillis | Long | 40000 | false | Maximum backoff (in milliseconds) between retries when creating secondary resources for Spark application. |
1212
| spark.kubernetes.operator.api.statusPatchMaxAttempts | Long | 3 | false | Maximal number of retry attempts of requests to k8s server for resource status update. This would be performed on top of k8s client spark.kubernetes.operator.retry.maxAttempts to overcome potential conflicting update on the same SparkApplication. This should be positive number. |
1313
| spark.kubernetes.operator.dynamicConfig.enabled | Boolean | false | false | When enabled, operator would load config property overrides dynamically at runtime. The source of the overrides is controlled by spark.kubernetes.operator.dynamicConfig.source. |

‎spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConf.java‎

Lines changed: 18 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -570,10 +570,11 @@ public final class SparkOperatorConf {
570570
* Maximal number of retry attempts of requesting secondary resource for Spark application. This
571571
* would be performed on top of k8s client spark.kubernetes.operator.retry.maxAttempts to overcome
572572
* potential conflicting reconcile on the same SparkApplication, as well as transient API server
573-
* errors and network-level timeouts. Exponential backoff with jitter is applied before retrying
574-
* whenever the API server response carries a Retry-After hint, or on 409 (Conflict) and 429 (Too
575-
* Many Requests) responses. Other retryable errors (408/500/502/503/504 and network timeouts)
576-
* without a Retry-After hint are retried immediately. This should be positive number
573+
* errors and requests the API server never answered. Exponential backoff with jitter is applied
574+
* before retrying whenever the API server response carries a Retry-After hint, on 409 (Conflict)
575+
* and 429 (Too Many Requests) responses, and on a request that went unanswered, which carries no
576+
* response status and so can never supply a Retry-After hint. The remaining retryable errors
577+
* (408/500/502/503/504) are retried immediately. This should be positive number
577578
*/
578579
public static final ConfigOption<Long> API_SECONDARY_RESOURCE_CREATE_MAX_ATTEMPTS =
579580
ConfigOption.<Long>builder()
@@ -584,11 +585,13 @@ public final class SparkOperatorConf {
584585
+ "application. This would be performed on top of k8s client "
585586
+ "spark.kubernetes.operator.retry.maxAttempts to overcome potential "
586587
+ "conflicting reconcile on the same SparkApplication, as well as API "
587-
+ "server errors (408/500/502/503/504) and network-level timeouts. "
588-
+ "Exponential backoff with jitter is applied before retrying whenever the "
589-
+ "API server response carries a Retry-After hint, or on 409 (Conflict) and "
590-
+ "429 (Too Many Requests) responses. Other retryable errors without a "
591-
+ "Retry-After hint are retried immediately. This should be positive number.")
588+
+ "server errors (408/500/502/503/504) and requests the API server never "
589+
+ "answered. Exponential backoff with jitter is applied before retrying "
590+
+ "whenever the API server response carries a Retry-After hint, on 409 "
591+
+ "(Conflict) and 429 (Too Many Requests) responses, and on a request that "
592+
+ "went unanswered, which carries no response status and so can never supply "
593+
+ "a Retry-After hint. The remaining retryable errors are retried "
594+
+ "immediately. This should be positive number.")
592595
.typeParameterClass(Long.class)
593596
.defaultValue(3L)
594597
.build();
@@ -807,8 +810,9 @@ public final class SparkOperatorConf {
807810

808811
/**
809812
* Initial backoff (in milliseconds) between retries when creating secondary resources. Backoff
810-
* is applied whenever the API server response carries a Retry-After hint, or on 409 (Conflict)
811-
* and 429 (Too Many Requests) responses.
813+
* is applied whenever the API server response carries a Retry-After hint, on 409 (Conflict) and
814+
* 429 (Too Many Requests) responses, and on a request the API server never answered, which
815+
* carries no response status and so can never supply a Retry-After hint.
812816
*/
813817
public static final ConfigOption<Long> API_SECONDARY_RESOURCE_CREATE_INITIAL_BACKOFF_MILLIS =
814818
ConfigOption.<Long>builder()
@@ -818,8 +822,9 @@ public final class SparkOperatorConf {
818822
"Initial backoff (in milliseconds) between retries "
819823
+ "when creating secondary resources for Spark application. "
820824
+ "Backoff is applied whenever the API server response carries a "
821-
+ "Retry-After hint, or on 409 (Conflict) and "
822-
+ "429 (Too Many Requests) responses.")
825+
+ "Retry-After hint, on 409 (Conflict) and 429 (Too Many Requests) "
826+
+ "responses, and on a request the API server never answered, which "
827+
+ "carries no response status and so can never supply a Retry-After hint.")
823828
.typeParameterClass(Long.class)
824829
.defaultValue(1000L)
825830
.build();

‎spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/ReconcilerUtils.java‎

Lines changed: 66 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -31,10 +31,15 @@
3131
import static org.apache.spark.k8s.operator.utils.ModelUtils.buildOwnerReferenceTo;
3232
import static org.apache.spark.k8s.operator.utils.SparkExceptionUtils.isConflictForExistingResource;
3333

34+
import java.security.cert.CertPathBuilderException;
35+
import java.security.cert.CertPathValidatorException;
36+
import java.security.cert.CertificateException;
3437
import java.util.List;
3538
import java.util.Optional;
3639
import java.util.concurrent.TimeUnit;
3740

41+
import javax.net.ssl.SSLPeerUnverifiedException;
42+
3843
import com.fasterxml.jackson.core.JsonProcessingException;
3944
import com.fasterxml.jackson.databind.ObjectMapper;
4045
import io.fabric8.kubernetes.api.model.DeletionPropagation;
@@ -57,6 +62,20 @@
5762
@Slf4j
5863
public final class ReconcilerUtils {
5964

65+
/**
66+
* The code fabric8 reports whenever no response status was received. It is the sentinel of
67+
* {@link KubernetesClientException#getCode()}, not an HTTP status, and it covers three unrelated
68+
* cases: a connection that broke on its way to the API server, which carries the failure that
69+
* broke it; a rejection the client raised before sending the request, such as a missing name or
70+
* resource version, which carries no cause at all; and a response that arrived but could not be
71+
* parsed, which carries the parsing failure. Only the first heals on its own, see
72+
* {@link #brokeOnTheWayToTheApiServer}.
73+
*/
74+
private static final int NO_RESPONSE_CODE = -1;
75+
76+
/** Maximum number of links followed when looking for the cause that broke a request. */
77+
private static final int MAX_CAUSE_DEPTH = 10;
78+
6079
private ReconcilerUtils() {}
6180

6281
/**
@@ -144,7 +163,7 @@ public static <T extends HasMetadata> Optional<T> getOrCreateSecondaryResource(
144163
} else if (e.getCode() == Constants.HTTP_TOO_MANY_REQUESTS) {
145164
log.debug("Server returned 429 Too Many Requests, will retry with backoff");
146165
} else if (isTransientError(e) || e.getCode() == HTTP_INTERNAL_ERROR) {
147-
// GET to avoid duplicate create attempt for timeouts (0) and transient 5xx
166+
// GET to avoid a duplicate create for transport failures, timeouts and transient 5xx
148167
current = getResource(client, resource);
149168
if (current.isPresent()) {
150169
return current;
@@ -279,7 +298,8 @@ private static boolean shouldBackoffBeforeRetry(KubernetesClientException e) {
279298
return true;
280299
}
281300
return switch (e.getCode()) {
282-
case HTTP_CONFLICT, Constants.HTTP_TOO_MANY_REQUESTS -> true;
301+
// A broken connection needs the delay most, and never carries a Retry-After to ask for one.
302+
case HTTP_CONFLICT, Constants.HTTP_TOO_MANY_REQUESTS, NO_RESPONSE_CODE -> true;
283303
default -> false;
284304
};
285305
}
@@ -297,20 +317,59 @@ public static boolean isFirstAttempt(Context<?> context) {
297317
}
298318

299319
/**
300-
* Whether the given failure is transport level rather than a decision by the API server.
320+
* Whether the given failure is expected to clear without anyone acting on it, so that a caller
321+
* may wait it out rather than report it. A broken connection and an overloaded or proxied server
322+
* qualify; a rejection by a reachable API server and a client side rejection do not.
301323
*
302324
* @param e The failure to classify.
303-
* @return True if the request did not reach a healthy API server, false otherwise.
325+
* @return True if the failure is expected to clear on its own, false otherwise.
304326
*/
305327
public static boolean isTransientError(KubernetesClientException e) {
306-
// code 0 is fabric8's sentinel for network-level failures (timeouts, connection resets, etc.)
307328
return switch (e.getCode()) {
308-
case 0, HTTP_CLIENT_TIMEOUT, HTTP_BAD_GATEWAY,
309-
HTTP_UNAVAILABLE, HTTP_GATEWAY_TIMEOUT -> true;
329+
case NO_RESPONSE_CODE -> brokeOnTheWayToTheApiServer(e);
330+
case HTTP_CLIENT_TIMEOUT, HTTP_BAD_GATEWAY, HTTP_UNAVAILABLE, HTTP_GATEWAY_TIMEOUT -> true;
310331
default -> false;
311332
};
312333
}
313334

335+
/**
336+
* Whether the given failure left the request unanswered, so that asking again may yet work. It
337+
* is defined as the complement of the status-less failures that repeating cannot change: a
338+
* rejection the client raised before sending anything, which carries no cause at all; an answer
339+
* that arrived but could not be parsed; a certificate the peer could not prove; and the caller
340+
* being interrupted, which is a decision to stop rather than a failure. Anything else that
341+
* carries a cause counts as unanswered.
342+
*
343+
* <p>Naming what cannot work, rather than what can, keeps this from tracking the exception types
344+
* of whichever HTTP client is plugged in. A connection the peer closes mid-response is the case
345+
* that matters: the client in use reports it with a type of its own which is not even an {@link
346+
* java.io.IOException}, so any list of recognized connection failures would silently miss it.
347+
*
348+
* @param e The failure to inspect.
349+
* @return True if the request went unanswered, false otherwise.
350+
*/
351+
private static boolean brokeOnTheWayToTheApiServer(KubernetesClientException e) {
352+
// Bounded walk: a cyclic cause chain must not hang the reconciler. The marker is nested, since
353+
// fabric8 wraps the failure and its own HTTP client wraps it again.
354+
Throwable cause = e.getCause();
355+
for (int depth = 0; cause != null && depth < MAX_CAUSE_DEPTH; depth++) {
356+
// The trust failures are the ones fabric8 itself refuses to retry. Keying on them rather
357+
// than on the handshake that reported them keeps a handshake that merely timed out
358+
// retriable. An interruption is matched on InterruptedException rather than on
359+
// InterruptedIOException, which a plain read timeout also extends.
360+
if (cause instanceof JsonProcessingException
361+
|| cause instanceof CertificateException
362+
|| cause instanceof CertPathValidatorException
363+
|| cause instanceof CertPathBuilderException
364+
|| cause instanceof SSLPeerUnverifiedException
365+
|| cause instanceof InterruptedException) {
366+
return false;
367+
}
368+
cause = cause.getCause();
369+
}
370+
return e.getCause() != null;
371+
}
372+
314373
/**
315374
* Clones an object using JSON serialization and deserialization.
316375
*

‎spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStepTest.java‎

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@
3232
import static org.mockito.Mockito.verifyNoInteractions;
3333
import static org.mockito.Mockito.when;
3434

35+
import java.net.SocketException;
3536
import java.time.Duration;
3637
import java.time.Instant;
3738
import java.util.List;
@@ -989,6 +990,57 @@ void kueueTransientApiFailurePublishesNoEvent() {
989990
verifyNoInteractions(eventRecorder);
990991
}
991992

993+
@Test
994+
void kueueUnansweredRequestPublishesNoEvent() {
995+
AppInitStep appInitStep = new AppInitStep();
996+
SparkAppContext mockContext = mock(SparkAppContext.class);
997+
SparkAppStatusRecorder recorder = mock(SparkAppStatusRecorder.class);
998+
SparkApplication application = new SparkApplication();
999+
application.setMetadata(kueueApplicationMetadata);
1000+
// A request that never reached the API server carries no response code, so only its cause
1001+
// marks it as one to wait out rather than report.
1002+
KubernetesClient failingClient = mock(KubernetesClient.class, RETURNS_DEEP_STUBS);
1003+
when(failingClient.scheduling().v1().priorityClasses().list())
1004+
.thenReturn(new PriorityClassList());
1005+
when(failingClient.resource(any(Workload.class)).create())
1006+
.thenThrow(new KubernetesClientException("closed", new SocketException("reset")));
1007+
when(mockContext.getResource()).thenReturn(application);
1008+
when(mockContext.getClient()).thenReturn(failingClient);
1009+
when(mockContext.getEventRecorder()).thenReturn(eventRecorder);
1010+
1011+
ReconcileProgress progress = appInitStep.reconcile(mockContext, recorder);
1012+
1013+
Assertions.assertEquals(
1014+
ReconcileProgress.completeAndRequeueAfter(
1015+
KueueWorkloadUtils.STALE_WORKLOAD_REQUEUE_INTERVAL),
1016+
progress);
1017+
verifyNoInteractions(eventRecorder);
1018+
}
1019+
1020+
@Test
1021+
void kueueClientSideRejectionPublishesAnEvent() {
1022+
AppInitStep appInitStep = new AppInitStep();
1023+
SparkAppContext mockContext = mock(SparkAppContext.class);
1024+
SparkAppStatusRecorder recorder = mock(SparkAppStatusRecorder.class);
1025+
SparkApplication application = new SparkApplication();
1026+
application.setMetadata(kueueApplicationMetadata);
1027+
// A rejection raised before the request was sent shares the absent response code, but nothing
1028+
// clears it on its own, so it keeps its event and the default interval.
1029+
KubernetesClient failingClient = mock(KubernetesClient.class, RETURNS_DEEP_STUBS);
1030+
when(failingClient.scheduling().v1().priorityClasses().list())
1031+
.thenReturn(new PriorityClassList());
1032+
when(failingClient.resource(any(Workload.class)).create())
1033+
.thenThrow(new KubernetesClientException("resourceVersion cannot be null"));
1034+
when(mockContext.getResource()).thenReturn(application);
1035+
when(mockContext.getClient()).thenReturn(failingClient);
1036+
when(mockContext.getEventRecorder()).thenReturn(eventRecorder);
1037+
1038+
ReconcileProgress progress = appInitStep.reconcile(mockContext, recorder);
1039+
1040+
Assertions.assertEquals(ReconcileProgress.completeAndDefaultRequeue(), progress);
1041+
verify(eventRecorder).record(any(EventRecord.class));
1042+
}
1043+
9921044
@Test
9931045
void failedDriverLookupBeforeKueueAdmissionIsRetried() {
9941046
// A failed verification is not an answer either before the admission: requesting quota for a

0 commit comments

Comments
 (0)