@@ -145,6 +145,9 @@ jobs:
           - kubernetes-version: "1.37.0"
             mode: static
             test-group: pi-java25
+          - kubernetes-version: "1.37.0"
+            mode: events
+            test-group: suspend-events
           - kubernetes-version: "1.37.0"
             mode: kueue
             test-group: kueue
@@ -244,6 +247,21 @@ jobs:
         if: matrix.mode == 'selector'
         run: |
           chainsaw test --test-dir ./tests/e2e/${{ matrix.test-group }} --parallel 1
+      - name: Run Spark K8s Operator on K8S with Kubernetes Events Enabled
+        if: matrix.mode == 'events'
+        run: |
+          eval $(minikube docker-env)
+          ./gradlew buildDockerImage
+          helm install spark --create-namespace -f \
+          build-tools/helm/spark-kubernetes-operator/values.yaml -f \
+          tests/e2e/helm/events-config-values.yaml \
+          build-tools/helm/spark-kubernetes-operator/
+          helm test spark
+          minikube docker-env --unset
+      - name: Run E2E Test with Kubernetes Events Enabled
+        if: matrix.mode == 'events'
+        run: |
+          chainsaw test --test-dir ./tests/e2e/${{ matrix.test-group }} --parallel 1
       - name: Run Spark K8s Operator on K8S with Kueue Enabled
         if: matrix.mode == 'kueue'
         run: |
 
@@ -16,7 +16,7 @@
  | spark.kubernetes.operator.dynamicConfig.reloadIntervalSeconds | Long | 60 | false | Interval (in seconds) at which the dynamic config file is re-read. Used by the 'file' source. | 
  | spark.kubernetes.operator.dynamicConfig.selector | String | app.kubernetes.io/name=spark-kubernetes-operator,app.kubernetes.io/component=operator-dynamic-config-overrides | false | The selector str applied to dynamic config map. Used by the 'configMap' source. | 
  | spark.kubernetes.operator.dynamicConfig.source | String | configMap | false | Source of dynamic config overrides when spark.kubernetes.operator.dynamicConfig.enabled is true. Supported values: 'configMap' (default) watches a ConfigMap via a Kubernetes informer and requires RBAC to read ConfigMaps; 'file' periodically reloads a properties file mounted from a ConfigMap and requires no extra RBAC. | 
- | spark.kubernetes.operator.events.enabled | Boolean | false | true | When enabled, operator would publish Kubernetes Event objects into the namespace of the Spark resource they describe, making them visible to users of that namespace via 'kubectl describe' and 'kubectl get events'. Events are published when a resource transitions into a new state, as Warning for failure states and Normal for most other states, when the operator cannot reconcile or update the status of a resource, and when a resource queued by Kueue waits for, gets or fails to request the admission. These are Event resources in the core API group, unrelated to the internal events that trigger reconciliation. | 
+ | spark.kubernetes.operator.events.enabled | Boolean | false | true | When enabled, operator would publish Kubernetes Event objects into the namespace of the Spark resource they describe, making them visible to users of that namespace via 'kubectl describe' and 'kubectl get events'. Events are published when a resource transitions into a new state, as Warning for failure states and Normal for most other states, when the operator cannot reconcile or update the status of a resource, when a resource is held by spec.suspend, and when a resource queued by Kueue waits for, gets or fails to request the admission. These are Event resources in the core API group, unrelated to the internal events that trigger reconciliation. | 
  | spark.kubernetes.operator.events.excludedReasons | String |  | true | Comma-separated list of Java regular expressions of Kubernetes Event reasons that operator would not publish even when 'spark.kubernetes.operator.events.enabled' is true, e.g. 'RunningHealthy,RunningWithPartialCapacity' or 'Running.*'. Each expression must match the whole reason and is case-sensitive. An invalid expression matches only the reason identical to it. If empty, operator would publish events of all reasons. | 
  | spark.kubernetes.operator.health.probePort | Integer | 19091 | false | The port used for health/readiness check probe status. | 
  | spark.kubernetes.operator.health.sentinelExecutorPoolSize | Integer | 3 | false | Size of executor service in Sentinel Managers to check the health of sentinel resources. | 
@@ -52,6 +52,7 @@
  | spark.kubernetes.operator.reconciler.retry.initialIntervalSeconds | Integer | 5 | false | Initial interval(in seconds) of retries on unhandled controller errors. | 
  | spark.kubernetes.operator.reconciler.retry.intervalMultiplier | Double | 1.5 | false | Interval multiplier of retries on unhandled controller errors. Setting this to 1 for linear retry. | 
  | spark.kubernetes.operator.reconciler.retry.maxIntervalSeconds | Integer | -1 | false | Max interval(in seconds) of retries on unhandled controller errors. Set to non-positive for unlimited. | 
+ | spark.kubernetes.operator.reconciler.suspendHoldRequeueIntervalSeconds | Long | 1800 | true | Requeue interval (in seconds) at which a resource held by spec.suspend is reconciled, so that its SuspendHeld event is republished. The hold ends only when a user clears spec.suspend, which arrives as a watch event and reconciles right away, so nothing waits for this interval. It is deliberately much coarser than spark.kubernetes.operator.reconciler.intervalSeconds, since a suspended resource has nothing to observe while each republish costs a read and a write on the API server. Keep it below the '--event-ttl' of the API server (one hour by default), or the event expires between repeats and the hold stops being visible. | 
  | spark.kubernetes.operator.reconciler.terminationTimeoutSeconds | Integer | 30 | false | Grace period for operator shutdown before reconciliation threads are killed. | 
  | spark.kubernetes.operator.reconciler.trimStateTransitionHistoryEnabled | Boolean | true | true | When enabled, operator would trim state transition history when a new attempt starts, keeping previous attempt summary only. | 
  | spark.kubernetes.operator.watchedNamespaces | String | default | true | Comma-separated list of namespaces that the operator would be watching for Spark resources. If set to '*', operator would watch all namespaces. | 
 
@@ -76,17 +76,20 @@ In addition, the operator publishes the following `Warning` events.
 | `StatusUpdateFailed` | A status patch is rejected. Transport-level errors are skipped. |
 | `KueueAdmissionRequestFailed` | Creating, reading or deleting a stale Kueue `Workload` fails. Transport-level errors are skipped and retried every 5 seconds, while a persistent failure is retried with the default interval. |
 
-For a resource queued by [Kueue](spark_custom_resources.md#kueue), the operator also publishes the
-following `Normal` events, since the resource stays in its initializing state while it waits.
+For a resource held by [`spec.suspend`](spark_custom_resources.md#suspend) or queued by
+[Kueue](spark_custom_resources.md#kueue), the operator also publishes the following `Normal`
+events, since the resource stays in its initializing state without a state transition.
 
 | Reason | When |
 |---|---|
+| `SuspendHeld` | The resource is held by `spec.suspend`, so the driver (or master and worker) is not requested. It is republished every 30 minutes while the hold lasts (`spark.kubernetes.operator.reconciler.suspendHoldRequeueIntervalSeconds`), so a repeat bumps the `count` of the one event rather than creating another. Suspending a queued resource releases its Kueue `Workload`, and the message says so, since the `KueueAdmissionPending` event it was queued with outlives that `Workload`. |
 | `KueueAdmissionPending` | The Kueue `Workload` waits for the admission. It is republished while it waits, so a repeat bumps the `count` of the one event rather than creating another. |
 | `KueueAdmitted` | Kueue admitted the `Workload`, so the driver (or master and worker) is requested. |
 
 The `reason` values are stable, while the `message` values may change between releases. Note that
 Kubernetes retains events only for a limited time (one hour by default), so the resource status
-remains the source of truth.
+remains the source of truth. The exception is a resource held by `spec.suspend` or by Kueue, whose
+status does not report the hold; that is why its event is republished while the hold lasts.
 
 To reduce noise, set `spark.kubernetes.operator.events.excludedReasons` to a comma-separated list
 of Java regular expressions. The operator does not publish events whose `reason` fully matches any
 
@@ -536,8 +536,13 @@ StatefulSets. Setting it back to `false` resumes the regular lifecycle.
 
 `Submitted` here is the operator's in-memory view. For a valid resource created with
 `suspend: true`, the initial `Submitted` status is not persisted to the API server, so
-`kubectl get` shows an empty `Current State` until initialization resumes. An application held
-later, in `ScheduledToRestart`, keeps the status its previous attempt already wrote.
+`kubectl get` shows an empty `Current State` and no state transition events are published until
+initialization resumes. Instead, the `SuspendHeld` [event](configuration.md#kubernetes-events) is
+published when enabled. Since the status is not there to fall back on, it is republished every 30
+minutes by default while the hold lasts, so that it outlives the event retention of the API
+server. An application held later, in `ScheduledToRestart`, keeps the status its previous attempt
+already wrote and gets the same event, since that status says that a restart is due, not that the
+next attempt is withheld.
 
 ``` yaml
 apiVersion: spark.apache.org/v1
 
@@ -262,6 +262,35 @@ public final class SparkOperatorConf {
           .defaultValue(15)
           .build();
 
+  /**
+   * Requeue interval (in seconds) at which a resource held by {@code spec.suspend} is reconciled,
+   * so that its {@code SuspendHeld} event is republished. The hold ends only when a user clears
+   * {@code spec.suspend}, which arrives as a watch event and reconciles right away, so nothing
+   * waits for this interval. It is deliberately much coarser than {@link
+   * #RECONCILER_INTERVAL_SECONDS}, since a suspended resource has nothing to observe while each
+   * republish costs a read and a write on the API server. Keep it below the {@code --event-ttl} of
+   * the API server (one hour by default), or the event expires between repeats and the hold stops
+   * being visible.
+   */
+  public static final ConfigOption<Long> SUSPEND_HOLD_REQUEUE_INTERVAL_SECONDS =
+      ConfigOption.<Long>builder()
+          .key("spark.kubernetes.operator.reconciler.suspendHoldRequeueIntervalSeconds")
+          .enableDynamicOverride(true)
+          .description(
+              "Requeue interval (in seconds) at which a resource held by spec.suspend is "
+                  + "reconciled, so that its SuspendHeld event is republished. The hold ends only "
+                  + "when a user clears spec.suspend, which arrives as a watch event and "
+                  + "reconciles right away, so nothing waits for this interval. It is "
+                  + "deliberately much coarser than "
+                  + "spark.kubernetes.operator.reconciler.intervalSeconds, since a suspended "
+                  + "resource has nothing to observe while each republish costs a read and a "
+                  + "write on the API server. Keep it below the '--event-ttl' of the API server "
+                  + "(one hour by default), or the event expires between repeats and the hold "
+                  + "stops being visible.")
+          .typeParameterClass(Long.class)
+          .defaultValue(1800L)
+          .build();
+
   /**
    * When enabled, operator would trim state transition history when a new attempt starts, keeping
    * previous attempt summary only.
@@ -292,10 +321,10 @@ public final class SparkOperatorConf {
                   + "namespace via 'kubectl describe' and 'kubectl get events'. Events are "
                   + "published when a resource transitions into a new state, as Warning for "
                   + "failure states and Normal for most other states, when the operator "
-                  + "cannot reconcile or update the status of a resource, and when a resource "
-                  + "queued by Kueue waits for, gets or fails to request the admission. These "
-                  + "are Event resources in the core API group, unrelated to the internal "
-                  + "events that trigger reconciliation.")
+                  + "cannot reconcile or update the status of a resource, when a resource is "
+                  + "held by spec.suspend, and when a resource queued by Kueue waits for, gets "
+                  + "or fails to request the admission. These are Event resources in the core "
+                  + "API group, unrelated to the internal events that trigger reconciliation.")
           .typeParameterClass(Boolean.class)
           .defaultValue(false)
           .build();
 
@@ -97,11 +97,32 @@ public Optional<Pod> getDriverPod() {
    * <p>The informer cache is only used to find a candidate. Since the cache may still hold the
    * pre-deletion snapshot of a previous attempt's pod, the candidate is verified against the API
    * server and returned only if it exists there and is not terminating. If the verification fails
-   * because of an API error, the pod is considered absent for this reconciliation.
+   * because of an API error, the pod is considered absent for this reconciliation. Callers that
+   * must not act on that assumption use {@link #getCurrentAttemptDriverPodStrictly()} instead.
    *
    * @return An Optional containing the driver Pod of the current attempt, or empty if not found.
    */
   public Optional<Pod> getCurrentAttemptDriverPod() {
+    try {
+      return getCurrentAttemptDriverPodStrictly();
+    } catch (KubernetesClientException e) {
+      log.warn(
+          "Failed to verify driver pod {} against the API server, considering it absent.",
+          getDriverPodSpec().getMetadata().getName(),
+          e);
+      return Optional.empty();
+    }
+  }
+
+  /**
+   * Same as {@link #getCurrentAttemptDriverPod()}, but reports a failed verification instead of
+   * reading it as an absent pod. A caller that would otherwise conclude that no driver has been
+   * requested, and act on that, uses this so that an API error does not look like an answer.
+   *
+   * @return An Optional containing the driver Pod of the current attempt, or empty if not found.
+   * @throws KubernetesClientException if the candidate cannot be verified against the API server.
+   */
+  public Optional<Pod> getCurrentAttemptDriverPodStrictly() {
     List<Pod> driverPods =
         josdkContext
             .getSecondaryResourcesAsStream(Pod.class)
@@ -120,25 +141,17 @@ public Optional<Pod> getCurrentAttemptDriverPod() {
     if (driverPods.stream().noneMatch(p -> driverPodName.equals(p.getMetadata().getName()))) {
       return Optional.empty();
     }
-    try {
-      Pod livePod =
-          josdkContext
-              .getClient()
-              .pods()
-              .inNamespace(sparkApplication.getMetadata().getNamespace())
-              .withName(driverPodName)
-              .get();
-      if (livePod == null || livePod.getMetadata().getDeletionTimestamp() != null) {
-        return Optional.empty();
-      }
-      return Optional.of(livePod);
-    } catch (KubernetesClientException e) {
-      log.warn(
-          "Failed to verify driver pod {} against the API server, considering it absent.",
-          driverPodName,
-          e);
+    Pod livePod =
+        josdkContext
+            .getClient()
+            .pods()
+            .inNamespace(sparkApplication.getMetadata().getNamespace())
+            .withName(driverPodName)
+            .get();
+    if (livePod == null || livePod.getMetadata().getDeletionTimestamp() != null) {
       return Optional.empty();
     }
+    return Optional.of(livePod);
   }
 
   /**
 
@@ -357,22 +357,26 @@ public static boolean isAdmissionChanged(final Workload newWorkload, final Workl
   }
 
   /**
-   * Deletes the Workload of the given resource so that Kueue releases its quota. Failures are
-   * logged only, because the Workload is garbage collected with its owner anyway.
+   * Deletes the Workload of the given resource so that Kueue releases its quota, and reports
+   * whether one was there to delete. Failures are logged only, because the Workload is garbage
+   * collected with its owner anyway.
    *
    * @param client The KubernetesClient.
    * @param owner The SparkApplication or SparkCluster owning the Workload.
+   * @return True if a Workload was deleted, false if there was none or the deletion failed.
    */
-  public static void releaseWorkload(final KubernetesClient client, final HasMetadata owner) {
+  public static boolean releaseWorkload(final KubernetesClient client, final HasMetadata owner) {
     String name = KueueWorkloadFactory.getWorkloadName(owner);
     try {
-      client
+      return !client
           .resources(Workload.class)
           .inNamespace(owner.getMetadata().getNamespace())
           .withName(name)
-          .delete();
+          .delete()
+          .isEmpty();
     } catch (KubernetesClientException e) {
       log.warn("Failed to release the Kueue Workload {}.", name, e);
+      return false;
     }
   }
 
 
@@ -34,6 +34,7 @@
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.api.model.Pod;
+import io.fabric8.kubernetes.client.KubernetesClientException;
 import lombok.extern.slf4j.Slf4j;
 
 import org.apache.spark.k8s.operator.Constants;
@@ -69,13 +70,20 @@ public ReconcileProgress reconcile(
       return proceed();
     }
     SparkApplication app = context.getResource();
-    if (app.getSpec().isSuspend() && !isDriverRequested(context)) {
-      log.debug("Application is suspended, driver resources would not be requested.");
-      if (KueueWorkloadFactory.hasQueueName(app)) {
-        // A resource suspended while queued must not keep holding the Kueue quota.
-        KueueWorkloadUtils.releaseWorkload(context.getClient(), app);
+    if (app.getSpec().isSuspend()) {
+      Optional<Pod> currentAttemptDriverPod;
+      try {
+        currentAttemptDriverPod = context.getCurrentAttemptDriverPodStrictly();
+      } catch (KubernetesClientException e) {
+        // Whether a driver of this attempt is live is unknown, not answered. Holding would claim
+        // in an event that none was requested, and would do so for the whole suspend hold
+        // interval, so look again with the steady-state interval instead.
+        log.warn("Failed to verify the driver pod of a suspended application, will retry.", e);
+        return completeAndDefaultRequeue();
+      }
+      if (currentAttemptDriverPod.isEmpty()) {
+        return SuspendUtils.holdForSuspend(context, "driver");
       }
-      return completeAndDefaultRequeue();
     }
     if (app.getStatus().getPreviousAttemptSummary() != null) {
       Instant lastTransitionTime = Instant.parse(currentState.getLastTransitionTime());
 
@@ -66,12 +66,7 @@ public ReconcileProgress reconcile(
     // A cluster whose master StatefulSet already exists has been requested before (e.g. the status
     // update to RunningHealthy failed), so let it complete its initialization even if suspended.
     if (cluster.getSpec().isSuspend() && !isMasterRequested(context)) {
-      log.debug("Cluster is suspended, master and worker resources would not be requested.");
-      if (KueueWorkloadFactory.hasQueueName(cluster)) {
-        // A resource suspended while queued must not keep holding the Kueue quota.
-        KueueWorkloadUtils.releaseWorkload(context.getClient(), cluster);
-      }
-      return completeAndDefaultRequeue();
+      return SuspendUtils.holdForSuspend(context, "master and workers");
     }
     if (cluster.getStatus().getPreviousAttemptSummary() != null) {
       Instant lastTransitionTime = Instant.parse(currentState.getLastTransitionTime());