fix: address PR #6550 review — retry, validation, per-FV status

- Remove redundant get_provider(); callers use .provider property - Replace SparkSession monkey-patch with per-thread session binding - Smart retry: only 5xx/429; fail fast on 401/403 with RBAC hint - Retry ConfigMap + SparkApplication creation (transient K8s errors) - Validate env entries: require name + value or valueFrom (K8s EnvVar) - Independent per-FV job status via CompletedMaterializationJob - Exit 1 on any FV failure so SparkApp CR reflects partial failure - Cleanup logs include kubectl delete command for manual recovery - Lint: pragma allowlist on test fixture URL Signed-off-by: Aniket Paluskar <apaluska@redhat.com>
ntkathole · Jul 17, 2026 · Jun 24, 2026 · Jul 3, 2026 · Jul 3, 2026 · Jul 3, 2026
commit d3fc4c86a76fd9fd7a065e5fa9f6d6a1c9092baa
@@ -424,10 +424,6 @@ def _get_provider(self) -> Provider:
         # TODO: Bake self.repo_path into self.config so that we dont only have one interface to paths
-        # TODO: Bake self.repo_path into self.config so that we dont only have one interface to paths
+        # Use locks or atomic operations for state management
+        import threading
+        state_lock = threading.Lock()
+        
+        with state_lock:
+            for fv, job in zip(regular_fvs, jobs):
+                fv_status = job.status()
+
+                if fv_status == MaterializationJobStatus.ERROR:
+                    failed_fvs.append(fv)
+                    if first_error is None and job.error():
+                        first_error = job.error()
+                else:
+                    succeeded_fvs.append(fv)
+
+            if failed_fvs:
+                self._rollback_fv_states(failed_fvs, previous_states)
-        # TODO: Bake self.repo_path into self.config so that we dont only have one interface to paths
+        # Use locks or atomic operations for state management
+        import threading
+        state_lock = threading.Lock()
+        
+        with state_lock:
+            for fv, job in zip(regular_fvs, jobs):
+                fv_status = job.status()
+
+                if fv_status == MaterializationJobStatus.ERROR:
+                    failed_fvs.append(fv)
+                    if first_error is None and job.error():
+                        first_error = job.error()
+                else:
+                    succeeded_fvs.append(fv)
+
+            if failed_fvs:
+                self._rollback_fv_states(failed_fvs, previous_states)
         return self.provider
 
-    def get_provider(self) -> Provider:
-        """Public accessor for the provider instance."""
-        return self._get_provider()
-
     def _rollback_fv_states(
         self,
         feature_views: list,

@@ -29,7 +29,14 @@
 from .config import (
     SparkApplicationComputeEngineConfig,  # noqa: F401 — required for Feast config resolution
 )
-from .job import SparkApplicationMaterializationJob
+from .job import (
+    _MAX_RETRIES,
+    _RETRY_BACKOFF_BASE,
+    CompletedMaterializationJob,
+    SparkApplicationMaterializationJob,
+    _is_retryable,
+    _rbac_hint,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -139,32 +146,46 @@ def materialize(
         job_id = uuid.uuid4().hex[:8]
 
         try:
-            self._create_configmap(job_id, tasks)
+            self._create_with_retry(
+                lambda: self._create_configmap(job_id, tasks),
+                "ConfigMap",
+                job_id,
+            )
         except ApiException as e:
             job = SparkApplicationMaterializationJob(
                 job_id,
                 self.config.namespace,
                 self.custom_api,
-                error=Exception(f"ConfigMap creation failed: {e.reason}"),
+                error=Exception(
+                    f"ConfigMap creation failed: HTTP {e.status} {e.reason}."
+                    f"{_rbac_hint(e.status)}"
+                ),
             )
             return [job for _ in tasks]
 
         try:
             cr = self._build_spark_application_cr(job_id)
-            self.custom_api.create_namespaced_custom_object(
-                group="sparkoperator.k8s.io",
-                version="v1beta2",
-                namespace=self.config.namespace,
-                plural="sparkapplications",
-                body=cr,
+            self._create_with_retry(
+                lambda: self.custom_api.create_namespaced_custom_object(
+                    group="sparkoperator.k8s.io",
+                    version="v1beta2",
+                    namespace=self.config.namespace,
+                    plural="sparkapplications",
+                    body=cr,
+                ),
+                "SparkApplication",
+                job_id,
             )
         except ApiException as e:
             self._cleanup(job_id)
             job = SparkApplicationMaterializationJob(
                 job_id,
                 self.config.namespace,
                 self.custom_api,
-                error=Exception(f"SparkApplication creation failed: {e.reason}"),
+                error=Exception(
+                    f"SparkApplication creation failed: HTTP {e.status} {e.reason}."
+                    f"{_rbac_hint(e.status)}"
+                ),
             )
             return [job for _ in tasks]
 
@@ -201,10 +222,14 @@ def _build_per_fv_jobs(
         job_id: str,
         job: SparkApplicationMaterializationJob,
     ) -> List[MaterializationJob]:
-        """Read each FV's state from registry to determine per-FV success/failure.
+        """Build one independent job object per FV from registry state.
+
+        The driver pod calls ``apply_materialization`` for each FV it
+        successfully materializes, setting state to ``AVAILABLE_ONLINE``.
+        FVs still in ``MATERIALIZING`` were not processed.
 
-        The pod calls apply_materialization for each succeeded FV, which sets
-        state to AVAILABLE_ONLINE. FVs still in MATERIALIZING were not processed.
+        Each returned job is an independent object so that a failed
+        SparkApplication does not pollute the status of succeeded FVs.
         """
         if len(tasks) <= 1:
             return [job for _ in tasks]
@@ -213,7 +238,7 @@ def _build_per_fv_jobs(
         for task in tasks:
             fv = registry.get_feature_view(task.feature_view.name, task.project)
             if getattr(fv, "state", None) == FeatureViewState.AVAILABLE_ONLINE:
-                jobs.append(job)
+                jobs.append(CompletedMaterializationJob(job_id))
             else:
                 jobs.append(
                     SparkApplicationMaterializationJob(
@@ -222,7 +247,7 @@ def _build_per_fv_jobs(
                         self.custom_api,
                         error=Exception(
                             f"Feature view '{task.feature_view.name}' was not "
-                            f"materialized by the SparkApplication pod"
+                            f"materialized by SparkApplication feast-sa-{job_id}"
                         ),
                     )
                 )
@@ -270,9 +295,10 @@ def _build_spark_application_cr(self, job_id: str) -> dict:
         }
         for entry in self.config.env:
             name = entry.get("name", "")
-            value = entry.get("value", "")
-            if name and value:
-                driver_env_conf[f"spark.kubernetes.driverEnv.{name}"] = value
+            if name and "value" in entry and entry["value"] is not None:
+                driver_env_conf[f"spark.kubernetes.driverEnv.{name}"] = str(
+                    entry["value"]
+                )
 
         spec = {
             "type": "Python",
@@ -399,22 +425,61 @@ def _get_driver_logs(self, job_id: str, tail_lines: int = 50) -> Optional[str]:
             logger.warning(f"Could not retrieve driver logs for feast-sa-{job_id}")
         return None
 
+    @staticmethod
+    def _create_with_retry(fn, resource_kind: str, job_id: str):
+        """Call *fn* with exponential backoff on transient K8s API errors."""
+        last_exc = None
+        for attempt in range(_MAX_RETRIES):
+            try:
+                return fn()
+            except ApiException as e:
+                if not _is_retryable(e):
+                    raise
+                last_exc = e
+                wait = _RETRY_BACKOFF_BASE**attempt
+                logger.warning(
+                    f"{resource_kind} feast-sa-{job_id}: create attempt "
+                    f"{attempt + 1}/{_MAX_RETRIES} failed (HTTP {e.status}), "
+                    f"retrying in {wait}s"
+                )
+                time.sleep(wait)
+        raise last_exc  # type: ignore[misc]
+
     def _cleanup(self, job_id: str):
-        for fn in [
-            lambda: self.custom_api.delete_namespaced_custom_object(
-                "sparkoperator.k8s.io",
-                "v1beta2",
-                self.config.namespace,
-                "sparkapplications",
-                f"feast-sa-{job_id}",
+        """Best-effort delete of SparkApplication + ConfigMap.
+
+        The SparkApplication CR is also garbage-collected by the Spark Operator
+        after ``spec.timeToLiveSeconds`` (default 1h), so a failed delete here
+        is not a resource leak — just delayed cleanup.  The ConfigMap is ours
+        and not covered by operator TTL.
+        """
+        resources = [
+            (
+                "SparkApplication",
+                lambda: self.custom_api.delete_namespaced_custom_object(
+                    "sparkoperator.k8s.io",
+                    "v1beta2",
+                    self.config.namespace,
+                    "sparkapplications",
+                    f"feast-sa-{job_id}",
+                ),
             ),
-            lambda: self.core_v1.delete_namespaced_config_map(
-                f"feast-sa-{job_id}",
-                self.config.namespace,
+            (
+                "ConfigMap",
+                lambda: self.core_v1.delete_namespaced_config_map(
+                    f"feast-sa-{job_id}",
+                    self.config.namespace,
+                ),
             ),
-        ]:
+        ]
+        for kind, fn in resources:
             try:
                 fn()
             except ApiException as e:
                 if e.status != 404:
-                    logger.warning(f"Cleanup failed: {e.reason}")
+                    logger.warning(
+                        f"Cleanup of {kind} feast-sa-{job_id} failed "
+                        f"(HTTP {e.status}): {e.reason}. "
+                        f"Manual cleanup: kubectl delete {kind.lower()} "
+                        f"feast-sa-{job_id} -n {self.config.namespace}"
+                    )
@@ -49,6 +49,24 @@ class SparkApplicationComputeEngineConfig(FeastConfigBaseModel):
     node_selector: Optional[Dict[str, str]] = None
     tolerations: List[dict] = []
 
+    @staticmethod
+    def _validate_env_entry(index: int, entry: object) -> None:
+        """Validate a single K8s EnvVar dict (must have name + value or valueFrom)."""
+        if not isinstance(entry, dict):
+            raise ValueError(
+                f"env[{index}] must be a dict, got {type(entry).__name__}: {entry}"
+            )
+        if "name" not in entry:
+            raise ValueError(
+                f"env[{index}] is missing required 'name'. "
+                f"Each env entry must be a K8s EnvVar dict. Got: {entry}"
+            )
+        if "value" not in entry and "valueFrom" not in entry:
+            raise ValueError(
+                f"env[{index}] must set 'value' or 'valueFrom' "
+                f"(K8s EnvVar spec). Got: {entry}"
+            )
+
     @model_validator(mode="after")
     def _validate_config(self) -> "SparkApplicationComputeEngineConfig":
         if self.staging_location:
@@ -59,9 +77,5 @@ def _validate_config(self) -> "SparkApplicationComputeEngineConfig":
                 stacklevel=2,
             )
         for i, entry in enumerate(self.env):
-            if "name" not in entry:
-                raise ValueError(
-                    f"env[{i}] is missing required 'name' key. "
-                    f"Each env entry must have at least 'name' and 'value'. Got: {entry}"
-                )
+            self._validate_env_entry(i, entry)
         return self
@@ -12,8 +12,23 @@
 
 logger = logging.getLogger(__name__)
 
-_MAX_POLL_RETRIES = 3
+_MAX_RETRIES = 3
 _RETRY_BACKOFF_BASE = 2
+_RETRYABLE_STATUS_CODES = {429, 500, 502, 503, 504}
+
+
+def _is_retryable(exc: ApiException) -> bool:
+    return exc.status in _RETRYABLE_STATUS_CODES
+
+
+def _rbac_hint(status: int) -> str:
+    if status in (401, 403):
+        return (
+            " Check that the Feast server ServiceAccount has the required"
+            " Role/RoleBinding for sparkapplications and configmaps in this namespace."
+        )
+    return ""
+
 
 _STATE_MAP = {
     "": MaterializationJobStatus.WAITING,
@@ -34,6 +49,34 @@
 assert len(_STATE_MAP) == 14
 
 
+class CompletedMaterializationJob(MaterializationJob):
+    """Lightweight stub for a FV whose materialization already succeeded.
+
+    Used by ``_build_per_fv_jobs`` so that each FV gets an independent job
+    object.  Unlike ``SparkApplicationMaterializationJob``, this never polls
+    the K8s API — the outcome is already known from the registry state.
+    """
+
+    def __init__(self, job_id: str):
+        super().__init__()
+        self._job_id = job_id
+
+    def status(self) -> MaterializationJobStatus:
+        return MaterializationJobStatus.SUCCEEDED
+
+    def error(self) -> Optional[BaseException]:
+        return None
+
+    def should_be_retried(self) -> bool:
+        return False
+
+    def job_id(self) -> str:
+        return f"feast-sa-{self._job_id}"
+
+    def url(self) -> Optional[str]:
+        return None
+
+
 class SparkApplicationMaterializationJob(MaterializationJob):
     def __init__(
         self,
@@ -74,7 +117,7 @@ def status(self) -> MaterializationJobStatus:
     def _get_cr_with_retry(self) -> Optional[dict]:
         """Fetch SparkApplication CR with exponential backoff on transient errors."""
         last_exc = None
-        for attempt in range(_MAX_POLL_RETRIES):
+        for attempt in range(_MAX_RETRIES):
             try:
                 return self.custom_api.get_namespaced_custom_object(
                     group="sparkoperator.k8s.io",
@@ -89,15 +132,21 @@ def _get_cr_with_retry(self) -> Optional[dict]:
                         f"SparkApplication feast-sa-{self._job_id} not found"
                     )
                     return None
+                if not _is_retryable(e):
+                    self._error = Exception(
+                        f"Kubernetes API error polling feast-sa-{self._job_id}: "
+                        f"HTTP {e.status} {e.reason}.{_rbac_hint(e.status)}"
+                    )
+                    return None
                 last_exc = e
                 wait = _RETRY_BACKOFF_BASE**attempt
                 logger.warning(
-                    f"API poll attempt {attempt + 1}/{_MAX_POLL_RETRIES} failed "
+                    f"API poll attempt {attempt + 1}/{_MAX_RETRIES} failed "
                     f"(HTTP {e.status}), retrying in {wait}s"
                 )
                 time.sleep(wait)
         self._error = Exception(
-            f"Failed to poll SparkApplication after {_MAX_POLL_RETRIES} attempts: "
+            f"Failed to poll SparkApplication after {_MAX_RETRIES} attempts: "
             f"{last_exc.reason if last_exc else 'unknown'}"
         )
         return None