@@ -77,6 +77,7 @@ In addition, the operator publishes the following `Warning` events.
 | `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. |
+| `KueueResourceFlavorReadFailed` | Reading the `ResourceFlavor`s which Kueue assigned to an admitted `Workload` fails, so the node selector and the tolerations of the flavors cannot be applied. Transport-level errors are skipped and retried every 5 seconds, while a persistent failure is retried with the default interval. |
 
 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`
 
@@ -47,10 +47,11 @@ under the License.
   `spark.kubernetes.operator.kueue.workloadInformer.enabled`, so that the operator watches
   `Workload` resources and an admission starts the queued resource right away. Without the watch,
   an admission is noticed on the next periodic reconcile. The watch requires Kueue to be installed
-  before the operator starts; otherwise the operator reports itself unhealthy. The priority of a
-  `Workload` comes from the cluster-scoped `workloadpriorityclasses` and `priorityclasses`, which
-  are granted only by `clusterRole.create`. Without the permission, the operator logs a warning and
-  creates the `Workload` without priority. See
+  before the operator starts; otherwise the operator reports itself unhealthy. The cluster-scoped
+  `resourceflavors`, `workloadpriorityclasses` and `priorityclasses` are granted only by
+  `clusterRole.create`. Without the permission, the priority of a `Workload` is skipped with a
+  warning, while the `ResourceFlavor`s which Kueue assigned cannot be read at all, so an admitted
+  resource is held until the permission is granted. See
   [Kueue](spark_custom_resources.md#kueue) for how resources opt into queueing.
 
 ## Spark Application Namespaces
 
@@ -624,6 +624,18 @@ spec:
   memory request or limit on the `worker` container, or `SPARK_WORKER_CORES` and
   `SPARK_WORKER_MEMORY`: a worker with none of them advertises the whole node to its executors,
   well above the default the `Workload` requests.
+* Like Kueue built-in integrations, the `nodeLabels` and `tolerations` of the `ResourceFlavor`s
+  assigned to a pod set are added to the node selector and tolerations of its pods. The executor
+  pods get them through the executor pod template, which the operator creates if not set. A node
+  label that conflicts with the node selector of the pods fails the resource with
+  `SchedulingFailure`, and deletes the `Workload` unless its driver or master is running already,
+  whose pods still hold the quota it reserved. Only fixing the node selector or the flavor
+  resolves it: a restarted attempt requests the quota again and hits the same conflict, so an
+  application which restarts on `SchedulingFailure` should bound the attempts with
+  [`maxRestartOnSchedulingFailure`](#granular-restart-control). `ResourceFlavor`s are
+  cluster-scoped, so reading them needs the rules which `operatorRbac.kueue.enabled` grants
+  through the ClusterRole, hence `operatorRbac.clusterRole.create` as well. Until they can be
+  read, the resource is held and the read is retried.
 * `spec.suspend` takes precedence. A suspended resource does not get a `Workload`, and suspending
   a queued resource deletes its `Workload` to release the quota. The resource is queued again when
   it is resumed.
 
@@ -19,11 +19,14 @@
 
 package org.apache.spark.k8s.operator.context;
 
+import java.util.Map;
+
 import io.fabric8.kubernetes.client.KubernetesClient;
 import io.javaoperatorsdk.operator.api.event.ResourceEventRecorder;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
 
 import org.apache.spark.k8s.operator.BaseResource;
+import org.apache.spark.k8s.operator.kueue.KueuePodSetFlavor;
 
 /**
  * Base class for context objects.
@@ -35,6 +38,9 @@ public abstract class BaseContext<CR extends BaseResource<?, ?, ?, ?, ?>> {
   /** The JOSDK context of the reconciliation this context belongs to. */
   protected final Context<?> josdkContext;
 
+  /** The flavors which Kueue assigned to the pod sets, applied to the secondary resources. */
+  protected Map<String, KueuePodSetFlavor> kueuePodSetFlavors = Map.of();
+
   /**
    * Constructs a context over the given JOSDK reconciliation context.
    *
@@ -51,6 +57,26 @@ protected BaseContext(Context<?> josdkContext) {
    */
   public abstract CR getResource();
 
+  /**
+   * Sets the flavors which Kueue assigned to the pod sets of the resource, so that its secondary
+   * resources carry their node selectors and tolerations.
+   *
+   * @param kueuePodSetFlavors The KueuePodSetFlavor by the pod set name.
+   */
+  public void setKueuePodSetFlavors(Map<String, KueuePodSetFlavor> kueuePodSetFlavors) {
+    synchronized (this) {
+      this.kueuePodSetFlavors = kueuePodSetFlavors;
+      applyKueuePodSetFlavors();
+    }
+  }
+
+  /**
+   * Applies the flavors to the secondary resource spec which was built before them, e.g. to find
+   * the driver pod or to check whether the master exists. Called while holding the lock of this
+   * context, so that a reader sees either spec with its flavors.
+   */
+  protected abstract void applyKueuePodSetFlavors();
+
   /**
    * Returns the Kubernetes client associated with this context.
    *
 
@@ -179,12 +179,25 @@ private SparkAppResourceSpec getSecondaryResourceSpec() {
       if (secondaryResourceSpec == null) {
         secondaryResourceSpec =
             SparkAppResourceSpecFactory.buildResourceSpec(
-                sparkApplication, josdkContext.getClient(), submissionWorker);
+                sparkApplication, josdkContext.getClient(), submissionWorker, kueuePodSetFlavors);
       }
       return secondaryResourceSpec;
     }
   }
 
+  /**
+   * Drops the resource spec built before the flavors, since they reach the driver and the
+   * executors through the pod templates which the submission builds the resources from.
+   */
+  @Override
+  protected void applyKueuePodSetFlavors() {
+    // The caller holds this lock already, which is re-entrant, so that the field is guarded by
+    // the same monitor as every other access to it.
+    synchronized (this) {
+      secondaryResourceSpec = null;
+    }
+  }
+
   /**
    * Returns the SparkApplication resource associated with this context.
    *
 
@@ -31,6 +31,8 @@
 import org.apache.spark.k8s.operator.SparkCluster;
 import org.apache.spark.k8s.operator.SparkClusterResourceSpec;
 import org.apache.spark.k8s.operator.SparkClusterSubmissionWorker;
+import org.apache.spark.k8s.operator.kueue.KueuePodSetFlavor;
+import org.apache.spark.k8s.operator.kueue.KueueWorkloadFactory;
 import org.apache.spark.k8s.operator.reconciler.SparkClusterResourceSpecFactory;
 
 /**
@@ -65,11 +67,40 @@ private SparkClusterResourceSpec getSecondaryResourceSpec() {
       if (secondaryResourceSpec == null) {
         secondaryResourceSpec =
             SparkClusterResourceSpecFactory.buildResourceSpec(sparkCluster, submissionWorker);
+        applyKueuePodSetFlavors();
       }
       return secondaryResourceSpec;
     }
   }
 
+  /**
+   * Adds the flavors to the StatefulSets of the resource spec, which is not built again for them:
+   * unlike the driver and the executors, the master and worker pods are created from the pod
+   * templates of these StatefulSets. Applying the same flavors twice is a no-op, so the spec built
+   * with them is left as it is.
+   */
+  @Override
+  protected void applyKueuePodSetFlavors() {
+    // The caller holds this lock already, which is re-entrant, so that the field is guarded by
+    // the same monitor as every other access to it.
+    synchronized (this) {
+      if (secondaryResourceSpec == null) {
+        return;
+      }
+      applyKueuePodSetFlavor(
+          KueueWorkloadFactory.PODSET_MASTER, secondaryResourceSpec.getMasterStatefulSet());
+      applyKueuePodSetFlavor(
+          KueueWorkloadFactory.PODSET_WORKER, secondaryResourceSpec.getWorkerStatefulSet());
+    }
+  }
+
+  private void applyKueuePodSetFlavor(String podSetName, StatefulSet statefulSet) {
+    KueuePodSetFlavor flavor = kueuePodSetFlavors.get(podSetName);
+    if (flavor != null) {
+      flavor.applyTo(statefulSet.getSpec().getTemplate().getSpec());
+    }
+  }
+
   /**
    * Returns the SparkCluster resource associated with this context.
    *