feast-dev
diff --git a/‎sdk/python/feast/pyspark/launchers/standalone/__init__.py‎
Lines changed: 10 additions & 2 deletions b/‎sdk/python/feast/pyspark/launchers/standalone/__init__.py‎
Lines changed: 10 additions & 2 deletions
diff --git a/‎sdk/python/feast/pyspark/launchers/standalone/local.py‎
Lines changed: 17 additions & 10 deletions b/‎sdk/python/feast/pyspark/launchers/standalone/local.py‎
Lines changed: 17 additions & 10 deletions
diff --git a/‎sdk/python/tests/test_streaming_control_loop.py‎
Lines changed: 6 additions & 2 deletions b/‎sdk/python/tests/test_streaming_control_loop.py‎
Lines changed: 6 additions & 2 deletions
@@ -1,3 +1,11 @@
-from .local import StandaloneClusterLauncher, StandaloneClusterRetrievalJob
+from .local import (
+    StandaloneClusterLauncher,
+    StandaloneClusterRetrievalJob,
+    reset_job_cache,
+)
 
-__all__ = ["StandaloneClusterRetrievalJob", "StandaloneClusterLauncher"]
+__all__ = [
+    "StandaloneClusterRetrievalJob",
+    "StandaloneClusterLauncher",
+    "reset_job_cache",
+]
@@ -25,10 +25,12 @@
 
 
 class JobCache:
-    """In-memory cache of Spark jobs.
-
-    This is necessary since we can't query Spark jobs in local mode
+    """
+    A *global* in-memory cache of Spark jobs.
 
+    This is necessary since we can't easily keep track of running Spark jobs in local mode, since
+    there is no external state (unlike EMR and Dataproc which keep track of the running jobs for
+    us).
     """
 
     # Map of job_id -> spark job
@@ -75,7 +77,12 @@ def get_job_by_id(self, job_id: str) -> SparkJob:
             return self.job_by_id[job_id]
 
 
-job_cache = JobCache()
+global_job_cache = JobCache()
+
+
+def reset_job_cache():
+    global global_job_cache
+    global_job_cache = JobCache()
 
 
 def _find_free_port():
@@ -293,7 +300,7 @@ def historical_feature_retrieval(
             self.spark_submit(job_params),
             job_params.get_destination_path(),
         )
-        job_cache.add_job(job)
+        global_job_cache.add_job(job)
         return job
 
     def offline_to_online_ingestion(
@@ -307,7 +314,7 @@ def offline_to_online_ingestion(
             self.spark_submit(ingestion_job_params, ui_port),
             ui_port,
         )
-        job_cache.add_job(job)
+        global_job_cache.add_job(job)
         return job
 
     def start_stream_to_online_ingestion(
@@ -322,22 +329,22 @@ def start_stream_to_online_ingestion(
             ui_port,
             ingestion_job_params.get_job_hash(),
         )
-        job_cache.add_job(job)
+        global_job_cache.add_job(job)
         return job
 
     def stage_dataframe(self, df, event_timestamp_column: str):
         raise NotImplementedError
 
     def get_job_by_id(self, job_id: str) -> SparkJob:
-        return job_cache.get_job_by_id(job_id)
+        return global_job_cache.get_job_by_id(job_id)
 
     def list_jobs(self, include_terminated: bool) -> List[SparkJob]:
         if include_terminated is True:
-            return job_cache.list_jobs()
+            return global_job_cache.list_jobs()
         else:
             return [
                 job
-                for job in job_cache.list_jobs()
+                for job in global_job_cache.list_jobs()
                 if job.get_status()
                 in (SparkJobStatus.STARTING, SparkJobStatus.IN_PROGRESS)
             ]
@@ -16,7 +16,10 @@
 from feast.feature import Feature
 from feast.feature_table import FeatureTable
 from feast.job_service import ensure_stream_ingestion_jobs
-from feast.pyspark.launchers.standalone import StandaloneClusterLauncher
+from feast.pyspark.launchers.standalone import (
+    StandaloneClusterLauncher,
+    reset_job_cache,
+)
 from feast.value_type import ValueType
 from tests.feast_core_server import CoreServicer as MockCoreServicer
 
@@ -37,7 +40,6 @@ def mock_server(servicer, add_fn):
         server.stop(None)
 
 
-CORE_URL = "core.feast.example.com"
 SERVING_URL = "serving.example.com"
 
 
@@ -101,6 +103,8 @@ def _delete_ft(self, client: Client):
     def test_streaming_job_control_loop(self) -> None:
         """ Test streaming job control loop logic. """
 
+        reset_job_cache()
+
         core_servicer = MockCoreServicer()
 
         processes: List[subprocess.Popen] = []