Revert "fix: Initialize ThreadPoolExecutor in sql registry (#210)"

This reverts commit 7274460.

Author: zabarn

sdk/python/feast/infra/registry/sql.py

@@ -1,6 +1,4 @@
-import atexit
 import logging
-import threading
 import uuid
 from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime, timezone
@@ -287,11 +285,6 @@ def __init__(
         self.thread_pool_executor_worker_count = (
             registry_config.thread_pool_executor_worker_count
         )
-        if self.thread_pool_executor_worker_count > 0:
-            self._executor = ThreadPoolExecutor(
-                max_workers=self.thread_pool_executor_worker_count
-            )
-            atexit.register(self._exit_handler)
         self.purge_feast_metadata = registry_config.purge_feast_metadata
         # Sync feast_metadata to projects table
         # when purge_feast_metadata is set to True, Delete data from
@@ -990,17 +983,14 @@ def process_project(project: Project):
             r.infra.CopyFrom(self.get_infra(project_name).to_proto())
 
         projects_list = self.list_projects(allow_cache=False)
-        if self._executor:
-            logger.info(
-                f"Thread count before executor.map: {len(threading.enumerate())}"
-            )
-            self._executor.map(process_project, projects_list)
-            logger.info(
-                f"Thread count after executor.map: {len(threading.enumerate())}"
-            )
+        if self.thread_pool_executor_worker_count == 0:
+            for project in projects_list:
+                process_project(project)
         else:
-            for p in projects_list:
-                process_project(p)
+            with ThreadPoolExecutor(
+                max_workers=self.thread_pool_executor_worker_count
+            ) as executor:
+                executor.map(process_project, projects_list)
 
         if last_updated_timestamps:
             r.last_updated.FromDatetime(max(last_updated_timestamps))
@@ -1427,10 +1417,3 @@ def get_project_metadata(
                             datetime.utcfromtimestamp(int(metadata_value))
                         )
         return project_metadata_model
-
-    def _exit_handler(self):
-        if self._executor:
-            logger.info("Shutting down SqlRegistry's ThreadPoolExecutor...")
-            self._executor.shutdown(wait=False, cancel_futures=True)
-            logger.info("ThreadPoolExecutor shut down successfully.")
-            self._executor = None