@@ -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, when a resource is held by spec.suspend, when a resource queued by Kueue waits for, gets or fails to request the admission, when the Kueue queue name label of a resource is ignored, and when a Kueue eviction of a running resource is not acted on. 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, and on other occasions, e.g. when an operation on a resource fails, or while a resource is held by spec.suspend or Kueue. See the 'Kubernetes Events' section of docs/configuration.md for all reasons and the required RBAC permissions. 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.events.minIntervalSeconds | Long | 300 | true | Minimum interval (in seconds) between two Kubernetes Events with the same reason and the same message on the same Spark resource. The first event of a reason on a resource is always published right away, only a repeat that says exactly what the last one said is dropped. Such a repeat only bumps the 'count' of the one Event object, so it carries no information while it still costs a read and a write on the API server. A repeat whose message differs, for instance because it embeds the cause of a failure, is always published, since the operator rewrites the message of the existing Event. An event is only published when a reconciliation emits it, so the effective period is this interval rounded up to the next repeat. Keep this interval plus the interval at which the repeats are emitted within the '--event-ttl' of the API server (one hour by default), or a repeat lands after the previous event expired and the hold stops being visible. The repeats are emitted every 'spark.kubernetes.operator.reconciler.suspendHoldRequeueIntervalSeconds' for SuspendHeld and every 'spark.kubernetes.operator.reconciler.intervalSeconds' for KueueAdmissionPending, so with the defaults this option may be at most 1800 and 3480 seconds respectively. Unlike 'spark.kubernetes.operator.events.excludedReasons', which blocks a reason entirely, this only limits how often it is published. If zero or negative, operator would publish every event. | 
  | spark.kubernetes.operator.health.probePort | Integer | 19091 | false | The port used for health/readiness check probe status. | 
 
@@ -57,6 +57,19 @@ When `spark.kubernetes.operator.events.enabled` is `true`, the operator publishe
 `Event` objects about `SparkApplication` and `SparkCluster` resources into their namespaces. They
 are visible via `kubectl describe` and `kubectl get events`.
 
+Publishing them requires the `get`, `create` and `patch` permissions on `events` in the core API
+group in each namespace of the Spark resources. Without them, the operator logs a
+`Could not record ... event` warning for each event that it cannot publish, and keeps reconciling.
+The operator ClusterRole and Roles created by the Helm chart include them. If you manage the RBAC
+of the operator yourself, e.g. with a custom `operatorRbac.roleBinding.roleRef`, add the following
+rule to its ClusterRole, or to its Role in each of those namespaces.
+
+```yaml
+- apiGroups: [""]
+  resources: ["events"]
+  verbs: ["get", "create", "patch"]
+```
+
 Every event is reported by `spark-kubernetes-operator`, which is set as both `source.component` and
 `reportingComponent`, so `kubectl describe` shows it in the `From` column and both the `source` and
 the `reportingComponent` field selectors match it. For example, the following lists the warnings
 
@@ -273,6 +273,14 @@ public final class Constants {
   public static final String CLUSTER_REQUEUED_MESSAGE =
       "Cluster is submitted again after its eviction by Kueue.";
 
+  /**
+   * Message indicating that the cluster failed after SchedulingFailure, whose message tells the
+   * cause.
+   */
+  public static final String CLUSTER_FAILED_AFTER_SCHEDULING_FAILURE_MESSAGE =
+      "Cluster failed, since requesting its resources failed. See the preceding "
+          + "SchedulingFailure state for the cause.";
+
   /** Message indicating that the cluster status cannot be processed. */
   public static final String UNKNOWN_CLUSTER_STATE_MESSAGE = "Cannot process cluster status.";
 
 
@@ -332,12 +332,11 @@ public final class SparkOperatorConf {
                   + "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, when a resource queued by Kueue waits for, gets or "
-                  + "fails to request the admission, when the Kueue queue name label of a "
-                  + "resource is ignored, and when a Kueue eviction of a running resource is not "
-                  + "acted on. These are Event resources in the core "
+                  + "failure states and Normal for most other states, and on other occasions, "
+                  + "e.g. when an operation on a resource fails, or while a resource is held by "
+                  + "spec.suspend or Kueue. See the 'Kubernetes Events' section of "
+                  + "docs/configuration.md for all reasons and the required RBAC permissions. "
+                  + "These are Event resources in the core "
                   + "API group, unrelated to the internal events that trigger reconciliation.")
           .typeParameterClass(Boolean.class)
           .defaultValue(false)
 
@@ -39,13 +39,19 @@ public final class ClusterUnknownStateStep extends ClusterReconcileStep {
   @Override
   public ReconcileProgress reconcile(
       SparkClusterContext context, SparkClusterStatusRecorder statusRecorder) {
+    ClusterStateSummary currentStateSummary =
+        context.getResource().getStatus().getCurrentState().getCurrentStateSummary();
     // Appending Failed again would write a new status on every reconcile, forever.
-    if (context.getResource().getStatus().getCurrentState().getCurrentStateSummary()
-        == ClusterStateSummary.Failed) {
+    if (currentStateSummary == ClusterStateSummary.Failed) {
       return ReconcileProgress.completeAndNoRequeue();
     }
-    ClusterState state =
-        new ClusterState(ClusterStateSummary.Failed, Constants.UNKNOWN_CLUSTER_STATE_MESSAGE);
+    // SchedulingFailure already reports the cause, so Failed points back to it rather than reading
+    // as another, unexplained failure.
+    String message =
+        currentStateSummary == ClusterStateSummary.SchedulingFailure
+            ? Constants.CLUSTER_FAILED_AFTER_SCHEDULING_FAILURE_MESSAGE
+            : Constants.UNKNOWN_CLUSTER_STATE_MESSAGE;
+    ClusterState state = new ClusterState(ClusterStateSummary.Failed, message);
     statusRecorder.persistStatus(context, context.getResource().getStatus().appendNewState(state));
     return ReconcileProgress.completeAndImmediateRequeue();
   }
 
@@ -19,6 +19,7 @@
 
 package org.apache.spark.k8s.operator.reconciler;
 
+import static org.apache.spark.k8s.operator.Constants.CLUSTER_FAILED_AFTER_SCHEDULING_FAILURE_MESSAGE;
 import static org.apache.spark.k8s.operator.Constants.LABEL_SPARK_CLUSTER_NAME;
 import static org.apache.spark.k8s.operator.utils.TestUtils.setConfigKey;
 import static org.assertj.core.api.Assertions.assertThat;
@@ -261,6 +262,10 @@ void testReconcileFailedClusterDoesNotLoop() throws Exception {
       assertEquals(
           ClusterStateSummary.Failed,
           cluster.getStatus().getCurrentState().getCurrentStateSummary());
+      // Failed points back to SchedulingFailure, which reports the cause.
+      assertEquals(
+          CLUSTER_FAILED_AFTER_SCHEDULING_FAILURE_MESSAGE,
+          cluster.getStatus().getCurrentState().getMessage());
       int historySize = cluster.getStatus().getStateTransitionHistory().size();
       verify(mockRecorder, times(1)).persistStatus(any(), any());