You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2026/03/12 20:10:56 UTC
(spark-kubernetes-operator) branch main updated: [SPARK-55905] Exponential backoff for getOrCreateSecondaryResource attempt
This is an automated email from the ASF dual-hosted git repository.
dongjoon 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 c099104 [SPARK-55905] Exponential backoff for getOrCreateSecondaryResource attempt
c099104 is described below
commit c0991045e5fcb1cc335c1652af2dbf38f5b36a7d
Author: Qi Tan <qi...@apple.com>
AuthorDate: Thu Mar 12 13:10:43 2026 -0700
[SPARK-55905] Exponential backoff for getOrCreateSecondaryResource attempt
### What changes were proposed in this pull request?
[SPARK-55905] Exponential backoff for getOrCreateSecondaryResource attempt
### Why are the changes needed?
Optimize Retry
### Does this PR introduce _any_ user-facing change?
yes
### How was this patch tested?
unit test
### Was this patch authored or co-authored using generative AI tooling?
no
Closes #542 from TQJADE/SPARK-55905.
Lead-authored-by: Qi Tan <qi...@apple.com>
Co-authored-by: Qi Tan <16...@users.noreply.github.com>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
docs/config_properties.md | 6 +-
.../org/apache/spark/k8s/operator/Constants.java | 3 +
.../k8s/operator/config/SparkOperatorConf.java | 67 ++++++++++++++++-
.../spark/k8s/operator/utils/BackoffUtils.java | 75 +++++++++++++++++++
.../spark/k8s/operator/utils/ReconcilerUtils.java | 13 ++++
.../reconciler/SparkAppReconcilerTest.java | 85 ++++++++++++++++++++++
6 files changed, 245 insertions(+), 4 deletions(-)
diff --git a/docs/config_properties.md b/docs/config_properties.md
index 2edeeda..f7040a2 100644
--- a/docs/config_properties.md
+++ b/docs/config_properties.md
@@ -4,7 +4,11 @@
| --- | --- | --- | --- | --- |
| spark.kubernetes.operator.api.retryAttemptAfterSeconds | Long | 1 | false | Default time (in seconds) to wait till next request. This would be used if server does not set Retry-After in response. Setting this to non-positive number means immediate retry. |
| 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. |
- | 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. This should be positive number. |
+ | spark.kubernetes.operator.api.secondaryResourceCreateBackoffJitterMillis | Long | 500 | false | Jitter (in milliseconds) added to backoff delay between retries when creating secondary resources for Spark application. |
+ | spark.kubernetes.operator.api.secondaryResourceCreateBackoffMultiplier | Double | 2.0 | false | Backoff multiplier applied to each retry interval when creating secondary resources for Spark application. |
+ | spark.kubernetes.operator.api.secondaryResourceCreateInitialBackoffMillis | Long | 1000 | false | Initial backoff (in milliseconds) between retries when creating secondary resources for Spark application. Backoff is applied only on 409 (Conflict) and 429 (Too Many Requests) responses. |
+ | 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 on 409 (Confl [...]
+ | spark.kubernetes.operator.api.secondaryResourceCreateMaxBackoffMillis | Long | 40000 | false | Maximum backoff (in milliseconds) between retries when creating secondary resources for Spark application. |
| 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. |
| spark.kubernetes.operator.dynamicConfig.enabled | Boolean | false | false | When enabled, operator would use config map as source of truth for config property override. The config map need to be created in spark.kubernetes.operator.namespace, and labeled with operator name. |
| spark.kubernetes.operator.dynamicConfig.reconcilerParallelism | Integer | 1 | false | Parallelism for dynamic config reconciler. Unbounded pool would be used if set to non-positive number. |
diff --git a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/Constants.java b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/Constants.java
index 898286e..1934946 100644
--- a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/Constants.java
+++ b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/Constants.java
@@ -225,4 +225,7 @@ public class Constants {
/** Message indicating that the cluster status cannot be processed. */
public static final String UNKNOWN_CLUSTER_STATE_MESSAGE = "Cannot process cluster status.";
+
+ /** HTTP 429 Too Many Requests (not defined in {@link java.net.HttpURLConnection}). */
+ public static final int HTTP_TOO_MANY_REQUESTS = 429;
}
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConf.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConf.java
index 9b8c54f..51b2710 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConf.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConf.java
@@ -381,7 +381,10 @@ public final class SparkOperatorConf {
* 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 transient API server
- * errors and network-level timeouts. This should be positive number
+ * errors and network-level timeouts. Exponential backoff with jitter is applied before retrying
+ * on 409 (Conflict) and 429 (Too Many Requests) responses. Other retryable errors
+ * (408/500/502/503/504 and network timeouts) are retried immediately. This should be positive
+ * number
*/
public static final ConfigOption<Long> API_SECONDARY_RESOURCE_CREATE_MAX_ATTEMPTS =
ConfigOption.<Long>builder()
@@ -392,8 +395,10 @@ public final class SparkOperatorConf {
+ "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. This should "
- + "be positive number.")
+ + "server errors (408/500/502/503/504) and network-level timeouts. "
+ + "Exponential backoff with jitter is applied before retrying on "
+ + "409 (Conflict) and 429 (Too Many Requests) responses. Other retryable "
+ + "errors are retried immediately. This should be positive number.")
.typeParameterClass(Long.class)
.defaultValue(3L)
.build();
@@ -592,6 +597,62 @@ public final class SparkOperatorConf {
.defaultValue(5)
.build();
+ /**
+ * Initial backoff (in milliseconds) between retries when creating secondary resources. Backoff
+ * is applied only on 409 (Conflict) and 429 (Too Many Requests) responses.
+ */
+ public static final ConfigOption<Long> API_SECONDARY_RESOURCE_CREATE_INITIAL_BACKOFF_MILLIS =
+ ConfigOption.<Long>builder()
+ .key("spark.kubernetes.operator.api.secondaryResourceCreateInitialBackoffMillis")
+ .enableDynamicOverride(false)
+ .description(
+ "Initial backoff (in milliseconds) between retries "
+ + "when creating secondary resources for Spark application. "
+ + "Backoff is applied only on 409 (Conflict) and "
+ + "429 (Too Many Requests) responses.")
+ .typeParameterClass(Long.class)
+ .defaultValue(1000L)
+ .build();
+
+ /** Maximum backoff (in milliseconds) between retries when creating secondary resources. */
+ public static final ConfigOption<Long> API_SECONDARY_RESOURCE_CREATE_MAX_BACKOFF_MILLIS =
+ ConfigOption.<Long>builder()
+ .key("spark.kubernetes.operator.api.secondaryResourceCreateMaxBackoffMillis")
+ .enableDynamicOverride(false)
+ .description(
+ "Maximum backoff (in milliseconds) between retries when creating "
+ + "secondary resources for Spark application.")
+ .typeParameterClass(Long.class)
+ .defaultValue(40000L)
+ .build();
+
+ /** Backoff multiplier applied to each retry interval when creating secondary resources. */
+ public static final ConfigOption<Double> API_SECONDARY_RESOURCE_CREATE_BACKOFF_MULTIPLIER =
+ ConfigOption.<Double>builder()
+ .key("spark.kubernetes.operator.api.secondaryResourceCreateBackoffMultiplier")
+ .enableDynamicOverride(false)
+ .description(
+ "Backoff multiplier applied to each retry interval when creating "
+ + "secondary resources for Spark application.")
+ .typeParameterClass(Double.class)
+ .defaultValue(2.0)
+ .build();
+
+ /**
+ * Jitter (in milliseconds) added to backoff delay between retries when creating secondary
+ * resources.
+ */
+ public static final ConfigOption<Long> API_SECONDARY_RESOURCE_CREATE_BACKOFF_JITTER_MILLIS =
+ ConfigOption.<Long>builder()
+ .key("spark.kubernetes.operator.api.secondaryResourceCreateBackoffJitterMillis")
+ .enableDynamicOverride(false)
+ .description(
+ "Jitter (in milliseconds) added to backoff delay between retries when "
+ + "creating secondary resources for Spark application.")
+ .typeParameterClass(Long.class)
+ .defaultValue(500L)
+ .build();
+
private SparkOperatorConf() {}
/**
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/BackoffUtils.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/BackoffUtils.java
new file mode 100644
index 0000000..553759b
--- /dev/null
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/BackoffUtils.java
@@ -0,0 +1,75 @@
+/*
+ * 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 org.apache.spark.k8s.operator.config.SparkOperatorConf.*;
+
+import java.util.concurrent.ThreadLocalRandom;
+
+import lombok.extern.slf4j.Slf4j;
+
+/** Utility class for exponential backoff with jitter on retryable API errors. */
+@Slf4j
+public final class BackoffUtils {
+
+ private BackoffUtils() {}
+
+ /**
+ * Sleeps with jitter exponential backoff before retrying resource creation.
+ *
+ * @see <a href="https://aws.amazon.com/blogs/architecture/exponential-backoff-and-jitter/">
+ * Exponential Backoff and Jitter</a>
+ */
+ public static void backoffSleep(int responseCode, long attemptCount, long maxAttempts) {
+ long initialBackoffMs = API_SECONDARY_RESOURCE_CREATE_INITIAL_BACKOFF_MILLIS.getValue();
+ long maxBackoffMs = API_SECONDARY_RESOURCE_CREATE_MAX_BACKOFF_MILLIS.getValue();
+ double backoffMultiplier = API_SECONDARY_RESOURCE_CREATE_BACKOFF_MULTIPLIER.getValue();
+ long jitterMs = API_SECONDARY_RESOURCE_CREATE_BACKOFF_JITTER_MILLIS.getValue();
+ long actualDelay =
+ computeBackoffDelay(
+ initialBackoffMs, backoffMultiplier, attemptCount, maxBackoffMs, jitterMs);
+ log.info(
+ "Retrying resource creation with exponential backoff. "
+ + "Delay: {}ms, responseCode: {}, attempt: {}/{}",
+ actualDelay,
+ responseCode,
+ attemptCount,
+ maxAttempts);
+ try {
+ Thread.sleep(actualDelay);
+ } catch (InterruptedException ex) {
+ log.info("Backoff sleep interrupted, waking up early to retry resource creation");
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ static long computeBackoffDelay(
+ long initialBackoffMs,
+ double backoffMultiplier,
+ long attemptCount,
+ long maxBackoffMs,
+ long jitterMs) {
+ long exponentialDelay =
+ (long) (initialBackoffMs * Math.pow(backoffMultiplier, attemptCount - 2));
+ long cappedDelay = Math.min(exponentialDelay, maxBackoffMs);
+ long jitter = jitterMs > 0 ? ThreadLocalRandom.current().nextLong(0, jitterMs + 1) : 0;
+ return cappedDelay + jitter;
+ }
+}
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 b042f4b..0896b70 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
@@ -48,6 +48,7 @@ import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
import lombok.extern.slf4j.Slf4j;
import org.apache.spark.k8s.operator.BaseResource;
+import org.apache.spark.k8s.operator.Constants;
import org.apache.spark.k8s.operator.reconciler.ReconcileProgress;
/** Utility class for reconciler operations. */
@@ -132,6 +133,8 @@ public final class ReconcilerUtils {
return current;
}
}
+ } else if (e.getCode() == Constants.HTTP_TOO_MANY_REQUESTS) {
+ log.debug("Server returned 429 Too Many Requests, will retry with backoff");
} else if (isTransientError(e) || e.getCode() == HTTP_INTERNAL_ERROR) {
// GET to avoid duplicate create attempt for timeouts (0) and transient 5xx
current = getResource(client, resource);
@@ -145,6 +148,9 @@ public final class ReconcilerUtils {
log.error("Max Retries exceeded while trying to create resource");
throw e;
}
+ if (shouldBackoffBeforeRetry(e)) {
+ BackoffUtils.backoffSleep(e.getCode(), attemptCount, maxAttempts);
+ }
}
}
}
@@ -222,6 +228,13 @@ public final class ReconcilerUtils {
}
}
+ private static boolean shouldBackoffBeforeRetry(KubernetesClientException e) {
+ return switch (e.getCode()) {
+ case HTTP_CONFLICT, Constants.HTTP_TOO_MANY_REQUESTS -> true;
+ default -> false;
+ };
+ }
+
private static boolean isTransientError(KubernetesClientException e) {
// code 0 is fabric8's sentinel for network-level failures (timeouts, connection resets, etc.)
return switch (e.getCode()) {
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 f9820a9..5164101 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
@@ -19,6 +19,13 @@
package org.apache.spark.k8s.operator.reconciler;
+import static java.net.HttpURLConnection.HTTP_CONFLICT;
+import static org.apache.spark.k8s.operator.config.SparkOperatorConf.API_SECONDARY_RESOURCE_CREATE_BACKOFF_JITTER_MILLIS;
+import static org.apache.spark.k8s.operator.config.SparkOperatorConf.API_SECONDARY_RESOURCE_CREATE_BACKOFF_MULTIPLIER;
+import static org.apache.spark.k8s.operator.config.SparkOperatorConf.API_SECONDARY_RESOURCE_CREATE_INITIAL_BACKOFF_MILLIS;
+import static org.apache.spark.k8s.operator.config.SparkOperatorConf.API_SECONDARY_RESOURCE_CREATE_MAX_ATTEMPTS;
+import static org.apache.spark.k8s.operator.config.SparkOperatorConf.API_SECONDARY_RESOURCE_CREATE_MAX_BACKOFF_MILLIS;
+import static org.apache.spark.k8s.operator.utils.TestUtils.setConfigKey;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -27,13 +34,18 @@ 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.times;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.util.List;
import java.util.Optional;
import io.fabric8.kubernetes.api.model.Pod;
+import io.fabric8.kubernetes.api.model.PodBuilder;
import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.KubernetesClientException;
+import io.fabric8.kubernetes.client.dsl.NamespaceableResource;
import io.javaoperatorsdk.operator.api.reconciler.Context;
import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
import org.junit.jupiter.api.BeforeEach;
@@ -173,4 +185,77 @@ class SparkAppReconcilerTest {
utils.verifyNoInteractions();
}
}
+
+ @SuppressWarnings("unchecked")
+ private NamespaceableResource<Pod> mockSecondaryResource(Pod pod) {
+ NamespaceableResource<Pod> mockResource = mock(NamespaceableResource.class);
+ when(mockClient.resource(pod)).thenReturn(mockResource);
+ return mockResource;
+ }
+
+ private void setFastBackoff() {
+ setConfigKey(API_SECONDARY_RESOURCE_CREATE_MAX_ATTEMPTS, 3L);
+ setConfigKey(API_SECONDARY_RESOURCE_CREATE_INITIAL_BACKOFF_MILLIS, 5000L);
+ setConfigKey(API_SECONDARY_RESOURCE_CREATE_MAX_BACKOFF_MILLIS, 10000L);
+ setConfigKey(API_SECONDARY_RESOURCE_CREATE_BACKOFF_JITTER_MILLIS, 0L);
+ setConfigKey(API_SECONDARY_RESOURCE_CREATE_BACKOFF_MULTIPLIER, 2.0);
+ }
+
+ private void restoreBackoffDefaults(
+ long maxAttempts,
+ long initialBackoff,
+ long maxBackoff,
+ long jitter,
+ double multiplier) {
+ setConfigKey(API_SECONDARY_RESOURCE_CREATE_MAX_ATTEMPTS, maxAttempts);
+ setConfigKey(API_SECONDARY_RESOURCE_CREATE_INITIAL_BACKOFF_MILLIS, initialBackoff);
+ setConfigKey(API_SECONDARY_RESOURCE_CREATE_MAX_BACKOFF_MILLIS, maxBackoff);
+ setConfigKey(API_SECONDARY_RESOURCE_CREATE_BACKOFF_JITTER_MILLIS, jitter);
+ setConfigKey(API_SECONDARY_RESOURCE_CREATE_BACKOFF_MULTIPLIER, multiplier);
+ }
+
+ private Pod buildTestPod() {
+ return new PodBuilder()
+ .withNewMetadata()
+ .withName("test-pod")
+ .withNamespace("default")
+ .endMetadata()
+ .build();
+ }
+
+ @Test
+ void secondaryResourceCreationRetriesWithBackoffOnConflict() {
+ long savedMaxAttempts = API_SECONDARY_RESOURCE_CREATE_MAX_ATTEMPTS.getValue();
+ long savedInitialBackoff = API_SECONDARY_RESOURCE_CREATE_INITIAL_BACKOFF_MILLIS.getValue();
+ long savedMaxBackoff = API_SECONDARY_RESOURCE_CREATE_MAX_BACKOFF_MILLIS.getValue();
+ long savedJitter = API_SECONDARY_RESOURCE_CREATE_BACKOFF_JITTER_MILLIS.getValue();
+ double savedMultiplier = API_SECONDARY_RESOURCE_CREATE_BACKOFF_MULTIPLIER.getValue();
+ try {
+ setFastBackoff();
+ Pod pod = buildTestPod();
+ NamespaceableResource<Pod> mockResource = mockSecondaryResource(pod);
+ KubernetesClientException conflict =
+ new KubernetesClientException("conflict", HTTP_CONFLICT, null);
+ when(mockResource.get()).thenReturn(null);
+ when(mockResource.create()).thenThrow(conflict).thenReturn(pod);
+
+ long start = System.currentTimeMillis();
+ Optional<Pod> result = ReconcilerUtils.getOrCreateSecondaryResource(mockClient, pod);
+ long elapsed = System.currentTimeMillis() - start;
+
+ assertTrue(result.isPresent());
+ verify(mockResource, times(2)).create();
+ assertTrue(elapsed >= 5000L,
+ "Expected backoff delay of at least "
+ + "5000ms, but was: "
+ + elapsed + "ms");
+ } finally {
+ restoreBackoffDefaults(
+ savedMaxAttempts,
+ savedInitialBackoff,
+ savedMaxBackoff,
+ savedJitter,
+ savedMultiplier);
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org