Repository: GoogleCloudDataproc/dataproc-jupyter-plugin (language: python)

Issue:
Update version number for the next release



Discussion / hints:

Fix patch:
diff --git a/dataproc_jupyter_plugin/__init__.py b/dataproc_jupyter_plugin/__init__.py
index 0ebc4796..63e6c106 100644
--- a/dataproc_jupyter_plugin/__init__.py
+++ b/dataproc_jupyter_plugin/__init__.py
@@ -12,15 +12,14 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.from ._version import __version__
 import logging
-from google.cloud.jupyter_config.tokenrenewer import CommandTokenRenewer
 
-from kernels_mixer.kernelspecs import MixingKernelSpecManager
+from google.cloud.jupyter_config.tokenrenewer import CommandTokenRenewer
+from jupyter_server.services.sessions.sessionmanager import SessionManager
 from kernels_mixer.kernels import MixingMappingKernelManager
+from kernels_mixer.kernelspecs import MixingKernelSpecManager
 from kernels_mixer.websockets import DelegatingWebsocketConnection
 
-from jupyter_server.services.sessions.sessionmanager import SessionManager
-
-from .handlers import setup_handlers, configure_gateway_client_url, DataprocPluginConfig
+from .handlers import DataprocPluginConfig, configure_gateway_client_url, setup_handlers
 
 
 def _jupyter_labextension_paths():
diff --git a/dataproc_jupyter_plugin/commons/constants.py b/dataproc_jupyter_plugin/commons/constants.py
index 1833d03b..c76052be 100644
--- a/dataproc_jupyter_plugin/commons/constants.py
+++ b/dataproc_jupyter_plugin/commons/constants.py
@@ -12,27 +12,20 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-
-from dataproc_jupyter_plugin.services.gcpUrlService import gcp_service_url
-
 ENVIRONMENT_API = "https://composer.googleapis.com/v1"
 TAGS = "dataproc_jupyter_plugin"
 GCS = "gs://"
 CONTENT_TYPE = "application/json"
-
-
-dataproc_url = gcp_service_url("dataproc")
-dataproc_url = gcp_service_url("dataproc")
-compute_url = gcp_service_url(
-    "compute", default_url="https://compute.googleapis.com/compute/v1"
-)
-metastore_url = gcp_service_url("metastore")
-cloudkms_url = gcp_service_url("cloudkms")
-cloudresourcemanager_url = gcp_service_url("cloudresourcemanager")
-datacatalog_url = gcp_service_url("datacatalog")
-storage_url = gcp_service_url(
-    "storage", default_url="https://storage.googleapis.com/storage/v1/"
-)
-dataplex_url = gcp_service_url(
-    "dataplex", default_url="https://dataplex.googleapis.com/v1"
-)
+PACKAGE_NAME = "dataproc_jupyter_plugin"
+COMPOSER_SERVICE_NAME = "composer"
+DATAPROC_SERVICE_NAME = "dataproc"
+BIGQUERY_SERVICE_NAME = "bigquery"
+DATACATALOG_SERVICE_NAME = "datacatalog"
+CLOUDRESOURCEMANAGER_SERVICE_NAME = "cloudresourcemanager"
+STORAGE_SERVICE_NAME = "storage"
+COMPUTE_SERVICE_NAME = "compute"
+METASTORE_SERVICE_NAME = "metastore"
+CLOUDKMS_SERVICE_NAME = "cloudkms"
+COMPUTE_SERVICE_DEFAULT_URL = "https://compute.googleapis.com/compute/v1"
+STORAGE_SERVICE_DEFAULT_URL = "https://storage.googleapis.com/storage/v1/"
+WRAPPER_PAPPERMILL_FILE = "wrapper_papermill.py"
diff --git a/dataproc_jupyter_plugin/commons/gcloudOperations.py b/dataproc_jupyter_plugin/commons/gcloudOperations.py
deleted file mode 100644
index 64db12ef..00000000
--- a/dataproc_jupyter_plugin/commons/gcloudOperations.py
+++ /dev/null
@@ -1,45 +0,0 @@
-# Copyright 2024 Google LLC
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# https://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-
-from google.cloud.jupyter_config.config import (
-    gcp_credentials,
-    gcp_project,
-    gcp_project_number,
-    gcp_region,
-)
-
-
-class GetCachedCredentials:
-    def get_cached_credentials(self):
-        credentials = {
-            "project_id": "",
-            "project_number": 0,
-            "region_id": "",
-            "access_token": "",
-            "config_error": 0,
-            "login_error": 0,
-        }
-
-        try:
-            credentials["project_id"] = gcp_project()
-            credentials["region_id"] = gcp_region()
-            credentials["config_error"] = 0
-            credentials["access_token"] = gcp_credentials()
-            credentials["project_number"] = gcp_project_number()
-            return credentials
-        except Exception as ex:
-            self.log.exception(f"Error fetching credentials from gcloud ", ex)
-            credentials["config_error"] = 1
-            return credentials
diff --git a/dataproc_jupyter_plugin/controllers/airflow.py b/dataproc_jupyter_plugin/controllers/airflow.py
new file mode 100644
index 00000000..0647f63b
--- /dev/null
+++ b/dataproc_jupyter_plugin/controllers/airflow.py
@@ -0,0 +1,168 @@
+# Copyright 2024 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import json
+
+import tornado
+from jupyter_server.base.handlers import APIHandler
+
+from dataproc_jupyter_plugin import credentials
+from dataproc_jupyter_plugin.services import airflow
+
+
+class DagListController(APIHandler):
+    @tornado.web.authenticated
+    async def get(self):
+        try:
+            client = airflow.Client(await credentials.get_cached(), self.log)
+            composer_name = self.get_argument("composer")
+            dag_list = await client.list_jobs(composer_name)
+            self.finish(json.dumps(dag_list))
+        except Exception as e:
+            self.log.exception("Error fetching cluster list")
+            self.finish({"error": str(e)})
+
+
+class DagDeleteController(APIHandler):
+    @tornado.web.authenticated
+    async def get(self):
+        try:
+            client = airflow.Client(await credentials.get_cached(), self.log)
+            composer = self.get_argument("composer")
+            dag_id = self.get_argument("dag_id")
+            from_page = self.get_argument("from_page", default=None)
+            delete_response = await client.delete_job(composer, dag_id, from_page)
+            if delete_response == 0:
+                self.finish(json.dumps({"status": delete_response}))
+            else:
+                self.log.exception("Error deleting dag file")
+                self.finish(json.dumps({"status": delete_response}))
+        except Exception as e:
+            self.log.exception(f"Error deleting dag file: {str(e)}")
+            self.finish({"error": str(e)})
+
+
+class DagUpdateController(APIHandler):
+    @tornado.web.authenticated
+    async def get(self):
+        try:
+            client = airflow.Client(await credentials.get_cached(), self.log)
+            composer = self.get_argument("composer")
+            dag_id = self.get_argument("dag_id")
+            status = self.get_argument("status")
+            update_response = await client.update_job(composer, dag_id, status)
+            if update_response == 0:
+                self.finish({"status": 0})
+            else:
+                self.log.exception("Error updating status")
+                self.finish(json.dumps(update_response))
+        except Exception as e:
+            self.log.exception(f"Error updating status: {str(e)}")
+            self.finish({"error": str(e)})
+
+
+class DagRunController(APIHandler):
+    @tornado.web.authenticated
+    async def get(self):
+        try:
+            client = airflow.Client(await credentials.get_cached(), self.log)
+            composer_name = self.get_argument("composer")
+            dag_id = self.get_argument("dag_id")
+            start_date = self.get_argument("start_date")
+            offset = self.get_argument("offset")
+            end_date = self.get_argument("end_date")
+            dag_run_list = await client.list_dag_runs(
+                composer_name, dag_id, start_date, end_date, offset
+            )
+            self.finish(json.dumps(dag_run_list))
+        except Exception as e:
+            self.log.exception(f"Error fetching dag run list {str(e)}")
+            self.finish({"error": str(e)})
+
+
+class DagRunTaskController(APIHandler):
+    @tornado.web.authenticated
+    async def get(self):
+        try:
+            client = airflow.Client(await credentials.get_cached(), self.log)
+            composer_name = self.get_argument("composer")
+            dag_id = self.get_argument("dag_id")
+            dag_run_id = self.get_argument("dag_run_id")
+            dag_run_list = await client.list_dag_run_task(
+                composer_name, dag_id, dag_run_id
+            )
+            self.finish(json.dumps(dag_run_list))
+        except Exception as e:
+            self.log.exception(f"Error fetching dag run tasks: {str(e)}")
+            self.finish({"error": str(e)})
+
+
+class DagRunTaskLogsController(APIHandler):
+    @tornado.web.authenticated
+    async def get(self):
+        try:
+            client = airflow.Client(await credentials.get_cached(), self.log)
+            composer_name = self.get_argument("composer")
+            dag_id = self.get_argument("dag_id")
+            dag_run_id = self.get_argument("dag_run_id")
+            task_id = self.get_argument("task_id")
+            task_try_number = self.get_argument("task_try_number")
+            dag_run_list = await client.list_dag_run_task_logs(
+                composer_name, dag_id, dag_run_id, task_id, task_try_number
+            )
+            self.finish(json.dumps(dag_run_list))
+        except Exception as e:
+            self.log.exception(f"Error fetching dag run task logs: {str(e)}")
+            self.finish({"error": str(e)})
+
+
+class EditDagController(APIHandler):
+    @tornado.web.authenticated
+    async def get(self):
+        try:
+            client = airflow.Client(await credentials.get_cached(), self.log)
+            bucket_name = self.get_argument("bucket_name")
+            dag_id = self.get_argument("dag_id")
+            dag_details = await client.edit_jobs(dag_id, bucket_name)
+            self.finish(json.dumps(dag_details))
+        except Exception as e:
+            self.log.exception("Error getting dag details")
+            self.finish({"error": str(e)})
+
+
+class ImportErrorController(APIHandler):
+    @tornado.web.authenticated
+    async def get(self):
+        try:
+            client = airflow.Client(await credentials.get_cached(), self.log)
+            composer_name = self.get_argument("composer")
+            import_errors_list = await client.list_import_errors(composer_name)
+            self.finish(json.dumps(import_errors_list))
+        except Exception as e:
+            self.log.exception("Error fetching import error list")
+            self.finish({"error": str(e)})
+
+
+class TriggerDagController(APIHandler):
+    @tornado.web.authenticated
+    async def get(self):
+        try:
+            client = airflow.Client(await credentials.get_cached(), self.log)
+            dag_id = self.get_argument("dag_id")
+            composer = self.get_argument("composer")
+            trigger = await client.dag_trigger(dag_id, composer)
+            self.finish(json.dumps(trigger))
+        except Exception as e:
+            self.log.exception("Error triggering dag")
+            self.finish({"error": str(e)})
diff --git a/dataproc_jupyter_plugin/controllers/bigquery.py b/dataproc_jupyter_plugin/controllers/bigquery.py
index 8af1cf20..f1db8853 100644
--- a/dataproc_jupyter_plugin/controllers/bigquery.py
+++ b/dataproc_jupyter_plugin/controllers/bigquery.py
@@ -15,15 +15,12 @@
 
 import json
 
-from jupyter_server.base.handlers import APIHandler
 import tornado
-
-from google.cloud.jupyter_config import gcp_project
+from jupyter_server.base.handlers import APIHandler
 
 from dataproc_jupyter_plugin import credentials
 from dataproc_jupyter_plugin.services import bigquery
 
-
 # GCP project holding BigQuery public datasets.
 BQ_PUBLIC_DATASET_PROJECT_ID = "bigquery-public-data"
 
@@ -38,7 +35,7 @@ async def get(self):
             dataset_list = await client.list_datasets(page_token, project_id)
             self.finish(json.dumps(dataset_list))
         except Exception as e:
-            self.log.exception(f"Error fetching datasets")
+            self.log.exception("Error fetching datasets")
             self.finish({"error": str(e)})
 
 
@@ -53,7 +50,7 @@ async def get(self):
             table_list = await client.list_table(dataset_id, page_token, project_id)
             self.finish(json.dumps(table_list))
         except Exception as e:
-            self.log.exception(f"Error fetching datasets")
+            self.log.exception("Error fetching datasets")
             self.finish({"error": str(e)})
 
 
@@ -67,7 +64,7 @@ async def get(self):
             dataset_info = await client.list_dataset_info(dataset_id, project_id)
             self.finish(json.dumps(dataset_info))
         except Exception as e:
-            self.log.exception(f"Error fetching dataset information")
+            self.log.exception("Error fetching dataset information")
             self.finish({"error": str(e)})
 
 
@@ -82,7 +79,7 @@ async def get(self):
             table_info = await client.list_table_info(dataset_id, table_id, project_id)
             self.finish(json.dumps(table_info))
         except Exception as e:
-            self.log.exception(f"Error fetching table information")
+            self.log.exception("Error fetching table information")
             self.finish({"error": str(e)})
 
 
@@ -105,7 +102,7 @@ async def get(self):
             )
             self.finish(json.dumps(preview_data))
         except Exception as e:
-            self.log.exception(f"Error fetching preview data")
+            self.log.exception("Error fetching preview data")
             self.finish({"error": str(e)})
 
 
@@ -124,7 +121,7 @@ async def get(self):
             project_list = await bq_projects_list()
             self.finish(json.dumps(project_list))
         except Exception as e:
-            self.log.exception(f"Error fetching projects")
+            self.log.exception("Error fetching projects")
             self.finish({"error": str(e)})
 
 
@@ -142,5 +139,5 @@ async def get(self):
             )
             self.finish(json.dumps(search_data))
         except Exception as e:
-            self.log.exception(f"Error fetching search data")
+            self.log.exception("Error fetching search data")
             self.finish({"error": str(e)})
diff --git a/dataproc_jupyter_plugin/controllers/clusterController.py b/dataproc_jupyter_plugin/controllers/clusterController.py
deleted file mode 100644
index bd3f2e68..00000000
--- a/dataproc_jupyter_plugin/controllers/clusterController.py
+++ /dev/null
@@ -1,37 +0,0 @@
-# Copyright 2023 Google LLC
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# https://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-
-import json
-from dataproc_jupyter_plugin.commons.gcloudOperations import GetCachedCredentials
-from jupyter_server.base.handlers import APIHandler
-import tornado
-from dataproc_jupyter_plugin.services.clusterListService import ClusterListService
-
-
-class ClusterListController(APIHandler):
-    @tornado.web.authenticated
-    def get(self):
-        try:
-            page_token = self.get_argument("pageToken")
-            page_size = self.get_argument("pageSize")
-            cluster = ClusterListService()
-            credentials = GetCachedCredentials.get_cached_credentials(self.log)
-            cluster_list = cluster.list_clusters(
-                credentials, page_size, page_token, self.log
-            )
-            self.finish(json.dumps(cluster_list))
-        except Exception as e:
-            self.log.exception(f"Error fetching cluster list")
-            self.finish({"error": str(e)})
diff --git a/dataproc_jupyter_plugin/controllers/composerController.py b/dataproc_jupyter_plugin/controllers/composer.py
similarity index 68%
rename from dataproc_jupyter_plugin/controllers/composerController.py
rename to dataproc_jupyter_plugin/controllers/composer.py
index 1f178a41..bcac863d 100644
--- a/dataproc_jupyter_plugin/controllers/composerController.py
+++ b/dataproc_jupyter_plugin/controllers/composer.py
@@ -1,4 +1,4 @@
-# Copyright 2023 Google LLC
+# Copyright 2024 Google LLC
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -13,22 +13,21 @@
 # limitations under the License.
 
 import json
-from dataproc_jupyter_plugin.commons.gcloudOperations import GetCachedCredentials
-from jupyter_server.base.handlers import APIHandler
+
 import tornado
-from dataproc_jupyter_plugin.services.composerService import ComposerService
-from requests import HTTPError
+from jupyter_server.base.handlers import APIHandler
 
+from dataproc_jupyter_plugin import credentials
+from dataproc_jupyter_plugin.services import composer
 
-class ComposerListController(APIHandler):
+
+class EnvironmentListController(APIHandler):
     @tornado.web.authenticated
-    def get(self):
+    async def get(self):
         """Returns names of available composer environments"""
         try:
-            environments_manager = ComposerService()
-            credentials = GetCachedCredentials.get_cached_credentials(self.log)
-            environments = environments_manager.list_environments(credentials, self.log)
-
+            client = composer.Client(await credentials.get_cached(), self.log)
+            environments = await client.list_environments()
         except Exception as e:
             self.log.exception(f"Error fetching composer environments: {str(e)}")
             self.finish({"error": str(e)})
diff --git a/dataproc_jupyter_plugin/controllers/dagController.py b/dataproc_jupyter_plugin/controllers/dagController.py
deleted file mode 100644
index 4442389e..00000000
--- a/dataproc_jupyter_plugin/controllers/dagController.py
+++ /dev/null
@@ -1,106 +0,0 @@
-# Copyright 2023 Google LLC
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# https://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-
-import json
-import subprocess
-from dataproc_jupyter_plugin.commons.gcloudOperations import GetCachedCredentials
-from jupyter_server.base.handlers import APIHandler
-import tornado
-from dataproc_jupyter_plugin.services.dagListService import (
-    DagListService,
-    DagDeleteService,
-    DagUpdateService,
-)
-from dataproc_jupyter_plugin.commons.constants import TAGS
-
-
-class DagListController(APIHandler):
-    @tornado.web.authenticated
-    def get(self):
-        try:
-            dag = DagListService()
-            composer_name = self.get_argument("composer")
-            credentials = GetCachedCredentials.get_cached_credentials(self.log)
-            dag_list = dag.list_jobs(credentials, composer_name, TAGS, self.log)
-            self.finish(json.dumps(dag_list))
-        except Exception as e:
-            self.log.exception(f"Error fetching cluster list")
-            self.finish({"error": str(e)})
-            self.finish({"error": str(e)})
-
-
-class DagDownloadController(APIHandler):
-    @tornado.web.authenticated
-    def get(self):
-        try:
-            dag_id = self.get_argument("dag_id")
-            bucket_name = self.get_argument("bucket_name")
-            cmd = f"gsutil cp 'gs://{bucket_name}/dataproc-notebooks/{dag_id}/input_notebooks/*' ~/Downloads"
-            process = subprocess.Popen(
-                cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True
-            )
-            output, _ = process.communicate()
-            if process.returncode == 0:
-                self.finish({"status": 0})
-            else:
-                self.log.exception(f"Error downloading input notebook")
-                self.finish({"status": 1})
-        except Exception as e:
-            self.log.exception(f"Error downloading input notebook: {str(e)}")
-            self.finish({"error": str(e)})
-
-
-class DagDeleteController(APIHandler):
-    @tornado.web.authenticated
-    def get(self):
-        try:
-            dag = DagDeleteService()
-            composer = self.get_argument("composer")
-            dag_id = self.get_argument("dag_id")
-            from_page = self.get_argument("from_page", default=None)
-            credentials = GetCachedCredentials.get_cached_credentials(self.log)
-            delete_response = dag.delete_job(
-                credentials, composer, dag_id, from_page, self.log
-            )
-            if delete_response == 0:
-                self.finish(json.dumps({"status": delete_response}))
-            else:
-                self.log.exception(f"Error deleting dag file")
-                self.finish(json.dumps({"status": delete_response}))
-        except Exception as e:
-            self.log.exception(f"Error deleting dag file: {str(e)}")
-            self.finish({"error": str(e)})
-
-
-class DagUpdateController(APIHandler):
-    @tornado.web.authenticated
-    def get(self):
-        try:
-            dag = DagUpdateService()
-            composer = self.get_argument("composer")
-            dag_id = self.get_argument("dag_id")
-            status = self.get_argument("status")
-            credentials = GetCachedCredentials.get_cached_credentials(self.log)
-            update_response = dag.update_job(
-                credentials, composer, dag_id, status, self.log
-            )
-            if update_response == 0:
-                self.finish({"status": 0})
-            else:
-                self.log.exception(f"Error updating status")
-                self.finish(json.dumps(update_response))
-        except Exception as e:
-            self.log.exception(f"Error updating status: {str(e)}")
-            self.finish({"error": str(e)})
diff --git a/dataproc_jupyter_plugin/controllers/dagRunController.py b/dataproc_jupyter_plugin/controllers/dagRunController.py
deleted file mode 100644
index f7cb2c20..00000000
--- a/dataproc_jupyter_plugin/controllers/dagRunController.py
+++ /dev/null
@@ -1,94 +0,0 @@
-# Copyright 2023 Google LLC
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# https://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-
-import json
-from dataproc_jupyter_plugin.commons.gcloudOperations import GetCachedCredentials
-from jupyter_server.base.handlers import APIHandler
-import tornado
-from dataproc_jupyter_plugin.services.dagRunService import (
-    DagRunListService,
-    DagRunTaskListService,
-    DagRunTaskLogsListService,
-)
-
-
-class DagRunController(APIHandler):
-    @tornado.web.authenticated
-    def get(self):
-        try:
-            dag_run = DagRunListService()
-            composer_name = self.get_argument("composer")
-            dag_id = self.get_argument("dag_id")
-            start_date = self.get_argument("start_date")
-            offset = self.get_argument("offset")
-            end_date = self.get_argument("end_date")
-            credentials = GetCachedCredentials.get_cached_credentials(self.log)
-            dag_run_list = dag_run.list_dag_runs(
-                credentials,
-                composer_name,
-                dag_id,
-                start_date,
-                end_date,
-                offset,
-                self.log,
-            )
-            self.finish(json.dumps(dag_run_list))
-        except Exception as e:
-            self.log.exception(f"Error fetching dag run list {str(e)}")
-            self.finish({"error": str(e)})
-
-
-class DagRunTaskController(APIHandler):
-    @tornado.web.authenticated
-    def get(self):
-        try:
-            dag_run = DagRunTaskListService()
-            composer_name = self.get_argument("composer")
-            dag_id = self.get_argument("dag_id")
-            dag_run_id = self.get_argument("dag_run_id")
-            credentials = GetCachedCredentials.get_cached_credentials(self.log)
-            dag_run_list = dag_run.list_dag_run_task(
-                credentials, composer_name, dag_id, dag_run_id, self.log
-            )
-            self.finish(json.dumps(dag_run_list))
-        except Exception as e:
-            self.log.exception(f"Error fetching dag run tasks: {str(e)}")
-            self.finish({"error": str(e)})
-
-
-class DagRunTaskLogsController(APIHandler):
-    @tornado.web.authenticated
-    def get(self):
-        try:
-            dag_run = DagRunTaskLogsListService()
-            composer_name = self.get_argument("composer")
-            dag_id = self.get_argument("dag_id")
-            dag_run_id = self.get_argument("dag_run_id")
-            task_id = self.get_argument("task_id")
-            task_try_number = self.get_argument("task_try_number")
-            credentials = GetCachedCredentials.get_cached_credentials(self.log)
-            dag_run_list = dag_run.list_dag_run_task_logs(
-                credentials,
-                composer_name,
-                dag_id,
-                dag_run_id,
-                task_id,
-                task_try_number,
-                self.log,
-            )
-            self.finish(json.dumps(dag_run_list))
-        except Exception as e:
-            self.log.exception(f"Error fetching dag run task logs: {str(e)}")
-            self.finish({"error": str(e)})
diff --git a/dataproc_jupyter_plugin/controllers/runtimeController.py b/dataproc_jupyter_plugin/controllers/dataproc.py
similarity index 53%
rename from dataproc_jupyter_plugin/controllers/runtimeController.py
rename to dataproc_jupyter_plugin/controllers/dataproc.py
index 52fbfbac..992dad0a 100644
--- a/dataproc_jupyter_plugin/controllers/runtimeController.py
+++ b/dataproc_jupyter_plugin/controllers/dataproc.py
@@ -1,4 +1,4 @@
-# Copyright 2023 Google LLC
+# Copyright 2024 Google LLC
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -12,25 +12,37 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-
 import json
-from dataproc_jupyter_plugin.commons.gcloudOperations import GetCachedCredentials
-from jupyter_server.base.handlers import APIHandler
+
 import tornado
-from dataproc_jupyter_plugin.services.runtimeListService import RuntimeListService
+from jupyter_server.base.handlers import APIHandler
+
+from dataproc_jupyter_plugin import credentials
+from dataproc_jupyter_plugin.services import dataproc
+
+
+class ClusterListController(APIHandler):
+    @tornado.web.authenticated
+    async def get(self):
+        try:
+            page_token = self.get_argument("pageToken")
+            page_size = self.get_argument("pageSize")
+            client = dataproc.Client(await credentials.get_cached(), self.log)
+            cluster_list = await client.list_clusters(page_size, page_token)
+            self.finish(json.dumps(cluster_list))
+        except Exception as e:
+            self.log.exception("Error fetching cluster list")
+            self.finish({"error": str(e)})
 
 
 class RuntimeController(APIHandler):
     @tornado.web.authenticated
-    def get(self):
+    async def get(self):
         try:
             page_token = self.get_argument("pageToken")
             page_size = self.get_argument("pageSize")
-            runtime = RuntimeListService()
-            credentials = GetCachedCredentials.get_cached_credentials(self.log)
-            runtime_list = runtime.list_runtime(
-                credentials, page_size, page_token, self.log
-            )
+            client = dataproc.Client(await credentials.get_cached(), self.log)
+            runtime_list = await client.list_runtime(page_size, page_token)
             self.finish(json.dumps(runtime_list))
         except Exception as e:
             self.log.exception(f"Error fetching runtime template list: {str(e)}")
diff --git a/dataproc_jupyter_plugin/controllers/editDagController.py b/dataproc_jupyter_plugin/controllers/editDagController.py
deleted file mode 100644
index 6a4f5ff4..00000000
--- a/dataproc_jupyter_plugin/controllers/editDagController.py
+++ /dev/null
@@ -1,35 +0,0 @@
-# Copyright 2023 Google LLC
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# https://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-import json
-from dataproc_jupyter_plugin.commons.gcloudOperations import GetCachedCredentials
-from jupyter_server.base.handlers import APIHandler
-import tornado
-from dataproc_jupyter_plugin.services.editDagService import DagEditService
-from dataproc_jupyter_plugin.commons.constants import TAGS
-
-
-class EditDagController(APIHandler):
-    @tornado.web.authenticated
-    def get(self):
-        try:
-            dag = DagEditService()
-            bucket_name = self.get_argument("bucket_name")
-            dag_id = self.get_argument("dag_id")
-            credentials = GetCachedCredentials.get_cached_credentials(self.log)
-            dag_details = dag.edit_jobs(dag_id, bucket_name, credentials, self.log)
-            self.finish(json.dumps(dag_details))
-        except Exception as e:
-            self.log.exception(f"Error getting dag details")
-            self.finish({"error": str(e)})
diff --git a/dataproc_jupyter_plugin/controllers/downloadOutputController.py b/dataproc_jupyter_plugin/controllers/executor.py
similarity index 51%
rename from dataproc_jupyter_plugin/controllers/downloadOutputController.py
rename to dataproc_jupyter_plugin/controllers/executor.py
index fec6541b..45ab0b09 100644
--- a/dataproc_jupyter_plugin/controllers/downloadOutputController.py
+++ b/dataproc_jupyter_plugin/controllers/executor.py
@@ -1,4 +1,4 @@
-# Copyright 2023 Google LLC
+# Copyright 2024 Google LLC
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -12,26 +12,40 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-
 import json
-from jupyter_server.base.handlers import APIHandler
+
 import tornado
-from dataproc_jupyter_plugin import handlers
-from dataproc_jupyter_plugin.services.downloadOutputService import DownloadOutputService
+from jupyter_server.base.handlers import APIHandler
+
+from dataproc_jupyter_plugin import credentials
+from dataproc_jupyter_plugin.services import executor
+
+
+class ExecutorController(APIHandler):
+    @tornado.web.authenticated
+    async def post(self):
+        try:
+            input_data = self.get_json_body()
+            client = executor.Client(await credentials.get_cached(), self.log)
+            result = await client.execute(input_data)
+            self.finish(json.dumps(result))
+        except Exception as e:
+            self.log.exception(f"Error creating dag schedule: {str(e)}")
+            self.finish({"error": str(e)})
 
 
-class downloadOutputController(APIHandler):
+class DownloadOutputController(APIHandler):
     @tornado.web.authenticated
-    def get(self):
+    async def get(self):
         try:
-            download_dag = DownloadOutputService()
+            client = executor.Client(await credentials.get_cached(), self.log)
             bucket_name = self.get_argument("bucket_name")
             dag_id = self.get_argument("dag_id")
             dag_run_id = self.get_argument("dag_run_id")
-            download_status = download_dag.download_dag_output(
-                bucket_name, dag_id, dag_run_id, self.log
+            download_status = client.download_dag_output(
+                bucket_name, dag_id, dag_run_id
             )
             self.finish(json.dumps({"status": download_status}))
         except Exception as e:
-            self.log.exception(f"Error download output file")
+            self.log.exception("Error download output file")
             self.finish({"error": str(e)})
diff --git a/dataproc_jupyter_plugin/controllers/executorController.py b/dataproc_jupyter_plugin/controllers/executorController.py
deleted file mode 100644
index 030c9c03..00000000
--- a/dataproc_jupyter_plugin/controllers/executorController.py
+++ /dev/null
@@ -1,32 +0,0 @@
-# Copyright 2023 Google LLC
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# https://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-
-from dataproc_jupyter_plugin.commons.gcloudOperations import GetCachedCredentials
-from jupyter_server.base.handlers import APIHandler
-import tornado
-from dataproc_jupyter_plugin.services.executorService import ExecutorService
-
-
-class ExecutorController(APIHandler):
-    @tornado.web.authenticated
-    def post(self):
-        try:
-            input_data = self.get_json_body()
-            execute = ExecutorService()
-            credentials = GetCachedCredentials.get_cached_credentials(self.log)
-            execute.execute(credentials, input_data, self.log)
-        except Exception as e:
-            self.log.exception(f"Error creating dag schedule: {str(e)}")
-            self.finish({"error": str(e)})
diff --git a/dataproc_jupyter_plugin/controllers/importErrorController.py b/dataproc_jupyter_plugin/controllers/importErrorController.py
deleted file mode 100644
index 6bf13750..00000000
--- a/dataproc_jupyter_plugin/controllers/importErrorController.py
+++ /dev/null
@@ -1,36 +0,0 @@
-# Copyright 2023 Google LLC
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# https://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-
-import json
-from dataproc_jupyter_plugin.commons.gcloudOperations import GetCachedCredentials
-from jupyter_server.base.handlers import APIHandler
-import tornado
-from dataproc_jupyter_plugin.services.importErrorService import ImportErrorService
-
-
-class ImportErrorController(APIHandler):
-    @tornado.web.authenticated
-    def get(self):
-        try:
-            import_errors = ImportErrorService()
-            composer_name = self.get_argument("composer")
-            credentials = GetCachedCredentials.get_cached_credentials(self.log)
-            import_errors_list = import_errors.list_import_errors(
-                credentials, composer_name, self.log
-            )
-            self.finish(json.dumps(import_errors_list))
-        except Exception as e:
-            self.log.exception(f"Error fetching import error list")
-            self.finish({"error": str(e)})
diff --git a/dataproc_jupyter_plugin/controllers/triggerDagController.py b/dataproc_jupyter_plugin/controllers/triggerDagController.py
deleted file mode 100644
index ebe91844..00000000
--- a/dataproc_jupyter_plugin/controllers/triggerDagController.py
+++ /dev/null
@@ -1,35 +0,0 @@
-# Copyright 2023 Google LLC
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# https://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-
-import json
-from dataproc_jupyter_plugin.commons.gcloudOperations import GetCachedCredentials
-from jupyter_server.base.handlers import APIHandler
-import tornado
-from dataproc_jupyter_plugin.services.triggerDagService import TriggerDagService
-
-
-class TriggerDagController(APIHandler):
-    @tornado.web.authenticated
-    def get(self):
-        try:
-            trigger_dag = TriggerDagService()
-            dag_id = self.get_argument("dag_id")
-            composer = self.get_argument("composer")
-            credentials = GetCachedCredentials.get_cached_credentials(self.log)
-            trigger = trigger_dag.dag_trigger(credentials, dag_id, composer, self.log)
-            self.finish(json.dumps(trigger))
-        except Exception as e:
-            self.log.exception(f"Error triggering dag")
-            self.finish({"error": str(e)})
diff --git a/dataproc_jupyter_plugin/credentials.py b/dataproc_jupyter_plugin/credentials.py
index 6da5dfed..97d23fe2 100644
--- a/dataproc_jupyter_plugin/credentials.py
+++ b/dataproc_jupyter_plugin/credentials.py
@@ -13,8 +13,8 @@
 # limitations under the License.
 
 from google.cloud.jupyter_config.config import (
-    async_run_gcloud_subcommand,
     async_get_gcloud_config,
+    async_run_gcloud_subcommand,
 )
 
 
diff --git a/dataproc_jupyter_plugin/dagTemplates/wrapper_papermill.py b/dataproc_jupyter_plugin/dagTemplates/wrapper_papermill.py
index 9e934a32..aa75b8d7 100644
--- a/dataproc_jupyter_plugin/dagTemplates/wrapper_papermill.py
+++ b/dataproc_jupyter_plugin/dagTemplates/wrapper_papermill.py
@@ -25,10 +25,11 @@
 
 import argparse
 import re
-from google.cloud.storage.client import Client
+import sys
+
 import papermill as pm
 import yaml
-import sys
+from google.cloud.storage.client import Client
 
 
 def main():
diff --git a/dataproc_jupyter_plugin/handlers.py b/dataproc_jupyter_plugin/handlers.py
index 0bbd193a..459379d1 100644
--- a/dataproc_jupyter_plugin/handlers.py
+++ b/dataproc_jupyter_plugin/handlers.py
@@ -12,63 +12,32 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import datetime
-import inspect
 import json
-import re
 import subprocess
-import threading
-import time
 
-from jupyter_server.base.handlers import APIHandler
-from jupyter_server.serverapp import ServerApp
-from jupyter_server.utils import url_path_join
-from jupyter_server.utils import ensure_async
-from requests import HTTPError
-import tornado
-from traitlets import Bool, Undefined, Unicode
-from traitlets.config import SingletonConfigurable
 
+import tornado
 from google.cloud.jupyter_config.config import (
     async_run_gcloud_subcommand,
-    async_get_gcloud_config,
     clear_gcloud_cache,
     gcp_kernel_gateway_url,
     gcp_project_number,
     gcp_region,
 )
+from jupyter_server.base.handlers import APIHandler
+from jupyter_server.serverapp import ServerApp
+from jupyter_server.utils import url_path_join
+from traitlets import Bool, Undefined, Unicode
+from traitlets.config import SingletonConfigurable
 
 from dataproc_jupyter_plugin import credentials, urls
-from dataproc_jupyter_plugin.controllers import bigquery
-
-from dataproc_jupyter_plugin.controllers.clusterController import ClusterListController
-from dataproc_jupyter_plugin.controllers.composerController import (
-    ComposerListController,
-)
-from dataproc_jupyter_plugin.controllers.dagController import (
-    DagDeleteController,
-    DagDownloadController,
-    DagListController,
-    DagUpdateController,
-)
-from dataproc_jupyter_plugin.controllers.dagRunController import (
-    DagRunController,
-    DagRunTaskController,
-    DagRunTaskLogsController,
+from dataproc_jupyter_plugin.controllers import (
+    airflow,
+    bigquery,
+    composer,
+    dataproc,
+    executor,
 )
-from dataproc_jupyter_plugin.controllers.downloadOutputController import (
-    downloadOutputController,
-)
-from dataproc_jupyter_plugin.controllers.editDagController import EditDagController
-from dataproc_jupyter_plugin.controllers.executorController import ExecutorController
-from dataproc_jupyter_plugin.controllers.importErrorController import (
-    ImportErrorController,
-)
-from dataproc_jupyter_plugin.controllers.runtimeController import RuntimeController
-from dataproc_jupyter_plugin.controllers.triggerDagController import (
-    TriggerDagController,
-)
-
 
 _region_not_set_error = """GCP region not set in gcloud.
 
@@ -219,21 +188,20 @@ def full_path(name):
         "configuration": ConfigHandler,
         "getGcpServiceUrls": UrlHandler,
         "log": LogHandler,
-        "composerList": ComposerListController,
-        "dagRun": DagRunController,
-        "dagRunTask": DagRunTaskController,
-        "dagRunTaskLogs": DagRunTaskLogsController,
-        "clusterList": ClusterListController,
-        "runtimeList": RuntimeController,
-        "createJobScheduler": ExecutorController,
-        "dagList": DagListController,
-        "dagDownload": DagDownloadController,
-        "dagDelete": DagDeleteController,
-        "dagUpdate": DagUpdateController,
-        "editJobScheduler": EditDagController,
-        "importErrorsList": ImportErrorController,
-        "triggerDag": TriggerDagController,
-        "downloadOutput": downloadOutputController,
+        "composerList": composer.EnvironmentListController,
+        "dagRun": airflow.DagRunController,
+        "dagRunTask": airflow.DagRunTaskController,
+        "dagRunTaskLogs": airflow.DagRunTaskLogsController,
+        "clusterList": dataproc.ClusterListController,
+        "runtimeList": dataproc.RuntimeController,
+        "createJobScheduler": executor.ExecutorController,
+        "dagList": airflow.DagListController,
+        "dagDelete": airflow.DagDeleteController,
+        "dagUpdate": airflow.DagUpdateController,
+        "editJobScheduler": airflow.EditDagController,
+        "importErrorsList": airflow.ImportErrorController,
+        "triggerDag": airflow.TriggerDagController,
+        "downloadOutput": executor.DownloadOutputController,
         "bigQueryDataset": bigquery.DatasetController,
         "bigQueryTable": bigquery.TableController,
         "bigQueryDatasetInfo": bigquery.DatasetInfoController,
diff --git a/dataproc_jupyter_plugin/models/models.py b/dataproc_jupyter_plugin/models/models.py
index 518e23f8..c9d7e652 100644
--- a/dataproc_jupyter_plugin/models/models.py
+++ b/dataproc_jupyter_plugin/models/models.py
@@ -14,6 +14,7 @@
 
 
 from typing import Dict, List, Optional
+
 from pydantic import BaseModel
 
 
diff --git a/dataproc_jupyter_plugin/services/airflow.py b/dataproc_jupyter_plugin/services/airflow.py
new file mode 100644
index 00000000..2aaada93
--- /dev/null
+++ b/dataproc_jupyter_plugin/services/airflow.py
@@ -0,0 +1,323 @@
+# Copyright 2024 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import re
+import subprocess
+import urllib
+
+import requests
+
+from dataproc_jupyter_plugin import urls
+from dataproc_jupyter_plugin.commons.constants import (
+    COMPOSER_SERVICE_NAME,
+    CONTENT_TYPE,
+    STORAGE_SERVICE_DEFAULT_URL,
+    STORAGE_SERVICE_NAME,
+    TAGS,
+)
+
+
+class Client:
+    def __init__(self, credentials, log):
+        self.log = log
+        if not (
+            ("access_token" in credentials)
+            and ("project_id" in credentials)
+            and ("region_id" in credentials)
+        ):
+            self.log.exception("Missing required credentials")
+            raise ValueError("Missing required credentials")
+        self._access_token = credentials["access_token"]
+        self.project_id = credentials["project_id"]
+        self.region_id = credentials["region_id"]
+
+    def create_headers(self):
+        return {
+            "Content-Type": CONTENT_TYPE,
+            "Authorization": f"Bearer {self._access_token}",
+        }
+
+    async def get_airflow_uri(self, composer_name):
+        try:
+            composer_url = await urls.gcp_service_url(COMPOSER_SERVICE_NAME)
+            api_endpoint = f"{composer_url}v1/projects/{self.project_id}/locations/{self.region_id}/environments/{composer_name}"
+            response = requests.get(api_endpoint, headers=self.create_headers())
+            if response.status_code == 200:
+                resp = response.json()
+                airflow_uri = resp.get("config", {}).get("airflowUri", "")
+                bucket = resp.get("storageConfig", {}).get("bucket", "")
+                return airflow_uri, bucket
+        except Exception as e:
+            self.log.exception(f"Error getting airflow uri: {str(e)}")
+            print(f"Error: {e}")
+
+    async def list_jobs(self, composer_name):
+        airflow_uri, bucket = await self.get_airflow_uri(composer_name)
+        try:
+            api_endpoint = f"{airflow_uri}/api/v1/dags?tags={TAGS}"
+            response = requests.get(api_endpoint, headers=self.create_headers())
+            if response.status_code == 200:
+                resp = response.json()
+            return resp, bucket
+        except Exception as e:
+            self.log.exception(f"Error getting dag list: {str(e)}")
+            return {"error": str(e)}
+
+    async def delete_job(self, composer_name, dag_id, from_page):
+        airflow_uri, bucket = await self.get_airflow_uri(composer_name)
+        try:
+            api_endpoint = f"{airflow_uri}/api/v1/dags/{dag_id}"
+            if from_page == None:
+                response = requests.delete(api_endpoint, headers=self.create_headers())
+                self.log.info(response)
+            cmd = f"gsutil rm gs://{bucket}/dags/dag_{dag_id}.py"
+            process = subprocess.Popen(
+                cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True
+            )
+            output, _ = process.communicate()
+            if process.returncode == 0:
+                return 0
+            else:
+                self.log.exception("Error deleting dag")
+                return 1
+        except Exception as e:
+            self.log.exception(f"Error deleting dag: {str(e)}")
+            return {"error": str(e)}
+
+    async def update_job(self, composer_name, dag_id, status):
+        airflow_uri, bucket = await self.get_airflow_uri(composer_name)
+        try:
+            api_endpoint = f"{airflow_uri}/api/v1/dags/{dag_id}"
+            if status == "true":
+                data = {"is_paused": False}
+            else:
+                data = {"is_paused": True}
+            response = requests.patch(
+                api_endpoint, json=data, headers=self.create_headers()
+            )
+            if response.status_code == 200:
+                return 0
+            else:
+                self.log.exception("Error updating status")
+                return 1
+        except Exception as e:
+            self.log.exception(f"Error updating status: {str(e)}")
+            return {"error": str(e)}
+
+    async def list_dag_runs(self, composer_name, dag_id, start_date, end_date, offset):
+        airflow_uri, bucket = await self.get_airflow_uri(composer_name)
+        try:
+            api_endpoint = f"{airflow_uri}/api/v1/dags/{dag_id}/dagRuns?execution_date_gte={start_date}&execution_date_lte={end_date}&offset={offset}"
+            response = requests.get(api_endpoint, headers=self.create_headers())
+            if response.status_code == 200:
+                resp = response.json()
+            return resp
+        except Exception as e:
+            self.log.exception(f"Error fetching dag run list: {str(e)}")
+            return {"error": str(e)}
+
+    async def list_dag_run_task(self, composer_name, dag_id, dag_run_id):
+        airflow_uri, bucket = await self.get_airflow_uri(composer_name)
+        try:
+            api_endpoint = (
+                f"{airflow_uri}/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances"
+            )
+            response = requests.get(api_endpoint, headers=self.create_headers())
+            if response.status_code == 200:
+                resp = response.json()
+            return resp
+        except Exception as e:
+            self.log.exception(f"Error fetching dag run task list: {str(e)}")
+            return {"error": str(e)}
+
+    async def list_dag_run_task_logs(
+        self, composer_name, dag_id, dag_run_id, task_id, task_try_number
+    ):
+        airflow_uri, bucket = await self.get_airflow_uri(composer_name)
+        try:
+            api_endpoint = f"{airflow_uri}/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/logs/{task_try_number}"
+            response = requests.get(api_endpoint, headers=self.create_headers())
+            if response.status_code == 200:
+                resp = response.text
+                resp_to_json = {"content": resp}
+            return resp_to_json
+        except Exception as e:
+            self.log.exception(f"Error fetching dag run task logs: {str(e)}")
+            return {"error": str(e)}
+
+    async def get_dag_file(self, dag_id, bucket_name):
+        try:
+
+            file_path = f"dags/dag_{dag_id}.py"
+            encoded_path = urllib.parse.quote(file_path, safe="")
+            storage_url = await urls.gcp_service_url(
+                STORAGE_SERVICE_NAME, default_url=STORAGE_SERVICE_DEFAULT_URL
+            )
+            api_endpoint = f"{storage_url}b/{bucket_name}/o/{encoded_path}?alt=media"
+            headers = {
+                "Content-Type": CONTENT_TYPE,
+                "Authorization": f"Bearer {self._access_token}",
+                "X-Goog-User-Project": self.project_id,
+            }
+            response = requests.get(api_endpoint, headers=headers)
+            if response.status_code == 200:
+                self.log.info("Dag file response fetched")
+
+            return response.content
+        except Exception as e:
+            self.log.exception(f"Error reading dag file: {str(e)}")
+            return {"error": str(e)}
+
+    async def edit_jobs(self, dag_id, bucket_name):
+        try:
+            cluster_name = ""
+            serverless_name = ""
+            email_on_success = "False"
+            stop_cluster_check = "False"
+            mode_selected = "serverless"
+            time_zone = ""
+            pattern = r"parameters\s*=\s*'''(.*?)'''"
+            file_response = await self.get_dag_file(dag_id, bucket_name)
+            content_str = file_response.decode("utf-8")
+            file_content = re.sub(r"(?<!\\)\\(?!n)", "", content_str)
+
+            if file_content:
+                for line in file_content.split("\n"):
+                    if "input_notebook" in line:
+                        input_notebook = line.split("=")[-1].strip().strip("'\"")
+                        break
+
+                for line in file_content.split("\n"):
+                    match = re.search(pattern, file_content, re.DOTALL)
+                    if match:
+                        parameters_yaml = match.group(1)
+                        parameters_list = [
+                            line.strip()
+                            for line in parameters_yaml.split("\n")
+                            if line.strip()
+                        ]
+                    else:
+                        parameters_list = []
+
+                for line in file_content.split("\n"):
+                    if "email" in line:
+                        # Extract the email string from the line
+                        email_str = (
+                            line.split(":")[-1].strip().strip("'\"").replace(",", "")
+                        )
+                        # Use regular expression to extract email addresses
+                        email_list = re.findall(
+                            r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b",
+                            email_str,
+                        )
+                        # Remove quotes from email addresses
+                        email_list = [email.strip("'\"") for email in email_list]
+                        break
+                for line in file_content.split("\n"):
+                    if "cluster_name" in line:
+                        cluster_name = (
+                            line.split(":")[-1]
+                            .strip()
+                            .strip("'\"}")
+                            .split("'")[0]
+                            .strip()
+                        )  # Extract project_id from the line
+                    elif "submit_pyspark_job" in line:
+                        mode_selected = "cluster"
+                    elif "'retries'" in line:
+                        retries = line.split(":")[-1].strip().strip("'\"},")
+                        retry_count = int(
+                            retries.strip("'\"")
+                        )  # Extract retry_count from the line
+                    elif "retry_delay" in line:
+                        retry_delay = int(
+                            line.split("int('")[1].split("')")[0]
+                        )  # Extract retry_delay from the line
+                    elif "email_on_failure" in line:
+                        second_part = line.split(":")[1].strip()
+                        email_on_failure = second_part.split("'")[
+                            1
+                        ]  # Extract email_failure from the line
+                    elif "email_on_retry" in line:
+                        second_part = line.split(":")[1].strip()
+                        email_on_retry = second_part.split("'")[1]
+                    elif "email_on_success" in line:
+                        second_part = line.split(":")[1].strip()
+                        email_on_success = second_part.split("'")[1]
+                    elif "schedule_interval" in line:
+                        schedule_interval = (
+                            line.split("=")[-1]
+                            .strip()
+                            .strip("'\"")
+                            .split(",")[0]
+                            .rstrip("'\"")
+                        )  # Extract schedule_interval from the line
+                    elif "stop_cluster_check" in line:
+                        stop_cluster_check = line.split("=")[-1].strip().strip("'\"")
+                    elif "serverless_name" in line:
+                        serverless_name = line.split("=")[-1].strip().strip("'\"")
+                    elif "time_zone" in line:
+                        time_zone = line.split("=")[-1].strip().strip("'\"")
+
+                payload = {
+                    "input_filename": input_notebook,
+                    "parameters": parameters_list,
+                    "mode_selected": mode_selected,
+                    "cluster_name": cluster_name,
+                    "serverless_name": serverless_name,
+                    "retry_count": retry_count,
+                    "retry_delay": retry_delay,
+                    "email_failure": email_on_failure,
+                    "email_delay": email_on_retry,
+                    "email_success": email_on_success,
+                    "email": email_list,
+                    "schedule_value": schedule_interval,
+                    "stop_cluster": stop_cluster_check,
+                    "time_zone": time_zone,
+                }
+                return payload
+
+            else:
+                self.log.exception("No Dag file found")
+        except Exception as e:
+            self.log.exception(f"Error downloading dag file: {str(e)}")
+
+    async def list_import_errors(self, composer):
+        airflow_uri, bucket = await self.get_airflow_uri(composer)
+        try:
+            api_endpoint = (
+                f"{airflow_uri}/api/v1/importErrors?order_by=-import_error_id"
+            )
+            response = requests.get(api_endpoint, headers=self.create_headers())
+            if response.status_code == 200:
+                resp = response.json()
+            return resp
+        except Exception as e:
+            self.log.exception(f"Error fetching import error list: {str(e)}")
+            return {"error": str(e)}
+
+    async def dag_trigger(self, dag_id, composer):
+        airflow_uri, bucket = await self.get_airflow_uri(composer)
+        try:
+            api_endpoint = f"{airflow_uri}/api/v1/dags/{dag_id}/dagRuns"
+            body = {"conf": {}}
+            response = requests.post(
+                api_endpoint, headers=self.create_headers(), json=body
+            )
+            if response.status_code == 200:
+                resp = response.json()
+            return resp
+        except Exception as e:
+            self.log.exception(f"Error triggering dag: {str(e)}")
+            return {"error": str(e)}
diff --git a/dataproc_jupyter_plugin/services/bigquery.py b/dataproc_jupyter_plugin/services/bigquery.py
index 59f509bc..f5c146ed 100644
--- a/dataproc_jupyter_plugin/services/bigquery.py
+++ b/dataproc_jupyter_plugin/services/bigquery.py
@@ -16,7 +16,12 @@
 import requests
 
 from dataproc_jupyter_plugin import urls
-from dataproc_jupyter_plugin.commons.constants import CONTENT_TYPE
+from dataproc_jupyter_plugin.commons.constants import (
+    BIGQUERY_SERVICE_NAME,
+    CLOUDRESOURCEMANAGER_SERVICE_NAME,
+    CONTENT_TYPE,
+    DATACATALOG_SERVICE_NAME,
+)
 
 
 class Client:
@@ -27,7 +32,7 @@ def __init__(self, credentials, log):
             and ("project_id" in credentials)
             and ("region_id" in credentials)
         ):
-            self.log.exception(f"Missing required credentials")
+            self.log.exception("Missing required credentials")
             raise ValueError("Missing required credentials")
         self._access_token = credentials["access_token"]
         self.project_id = credentials["project_id"]
@@ -41,7 +46,8 @@ def create_headers(self):
 
     async def list_datasets(self, page_token, project_id):
         try:
-            api_endpoint = f"https://bigquery.googleapis.com/bigquery/v2/projects/{project_id}/datasets?pageToken={page_token}"
+            bigquery_url = await urls.gcp_service_url(BIGQUERY_SERVICE_NAME)
+            api_endpoint = f"{bigquery_url}bigquery/v2/projects/{project_id}/datasets?pageToken={page_token}"
             response = requests.get(api_endpoint, headers=self.create_headers())
             if response.status_code == 200:
                 resp = response.json()
@@ -49,12 +55,13 @@ async def list_datasets(self, page_token, project_id):
             else:
                 raise Exception(f"Error response from BigQuery: {response}")
         except Exception as e:
-            self.log.exception(f"Error fetching datasets list")
+            self.log.exception("Error fetching datasets list")
             return {"error": str(e)}
 
     async def list_table(self, dataset_id, page_token, project_id):
         try:
-            api_endpoint = f"https://bigquery.googleapis.com/bigquery/v2/projects/{project_id}/datasets/{dataset_id}/tables?pageToken={page_token}"
+            bigquery_url = await urls.gcp_service_url(BIGQUERY_SERVICE_NAME)
+            api_endpoint = f"{bigquery_url}bigquery/v2/projects/{project_id}/datasets/{dataset_id}/tables?pageToken={page_token}"
             response = requests.get(api_endpoint, headers=self.create_headers())
             if response.status_code == 200:
                 resp = response.json()
@@ -62,12 +69,15 @@ async def list_table(self, dataset_id, page_token, project_id):
             else:
                 raise Exception(f"Error response from BigQuery: {response}")
         except Exception as e:
-            self.log.exception(f"Error fetching tables list")
+            self.log.exception("Error fetching tables list")
             return {"error": str(e)}
 
     async def list_dataset_info(self, dataset_id, project_id):
         try:
-            api_endpoint = f"https://bigquery.googleapis.com/bigquery/v2/projects/{project_id}/datasets/{dataset_id}"
+            bigquery_url = await urls.gcp_service_url(BIGQUERY_SERVICE_NAME)
+            api_endpoint = (
+                f"{bigquery_url}bigquery/v2/projects/{project_id}/datasets/{dataset_id}"
+            )
             response = requests.get(api_endpoint, headers=self.create_headers())
             if response.status_code == 200:
                 resp = response.json()
@@ -75,12 +85,13 @@ async def list_dataset_info(self, dataset_id, project_id):
             else:
                 raise Exception(f"Error response from BigQuery: {response}")
         except Exception as e:
-            self.log.exception(f"Error fetching dataset info")
+            self.log.exception("Error fetching dataset info")
             return {"error": str(e)}
 
     async def list_table_info(self, dataset_id, table_id, project_id):
         try:
-            api_endpoint = f"https://bigquery.googleapis.com/bigquery/v2/projects/{project_id}/datasets/{dataset_id}/tables/{table_id}"
+            bigquery_url = await urls.gcp_service_url(BIGQUERY_SERVICE_NAME)
+            api_endpoint = f"{bigquery_url}bigquery/v2/projects/{project_id}/datasets/{dataset_id}/tables/{table_id}"
             response = requests.get(api_endpoint, headers=self.create_headers())
             if response.status_code == 200:
                 resp = response.json()
@@ -100,7 +111,8 @@ async def bigquery_preview_data(
         project_id,
     ):
         try:
-            api_endpoint = f"https://bigquery.googleapis.com/bigquery/v2/projects/{project_id}/datasets/{dataset_id}/tables/{table_id}/data?maxResults={max_results}&startIndex={start_index}"
+            bigquery_url = await urls.gcp_service_url(BIGQUERY_SERVICE_NAME)
+            api_endpoint = f"{bigquery_url}bigquery/v2/projects/{project_id}/datasets/{dataset_id}/tables/{table_id}/data?maxResults={max_results}&startIndex={start_index}"
             response = requests.get(api_endpoint, headers=self.create_headers())
             if response.status_code == 200:
                 resp = response.json()
@@ -108,12 +120,12 @@ async def bigquery_preview_data(
             else:
                 raise Exception(f"Error response from BigQuery: {response}")
         except Exception as e:
-            self.log.exception(f"Error fetching preview data")
+            self.log.exception("Error fetching preview data")
             return {"error": str(e)}
 
     async def bigquery_search(self, search_string, type, system, projects):
         try:
-            datacatalog_url = await urls.gcp_service_url("datacatalog")
+            datacatalog_url = await urls.gcp_service_url(DATACATALOG_SERVICE_NAME)
             api_endpoint = f"{datacatalog_url}v1/catalog:search"
             headers = {
                 "Content-Type": CONTENT_TYPE,
@@ -149,7 +161,10 @@ async def bigquery_search(self, search_string, type, system, projects):
 
     async def bigquery_projects(self, dataset_id, table_id):
         try:
-            api_endpoint = f"https://cloudresourcemanager.googleapis.com/v1/projects"
+            cloudresourcemanager_url = await urls.gcp_service_url(
+                CLOUDRESOURCEMANAGER_SERVICE_NAME
+            )
+            api_endpoint = f"{cloudresourcemanager_url}v1/projects"
             response = requests.get(api_endpoint, headers=self.create_headers())
             if response.status_code == 200:
                 resp = response.json()
@@ -157,5 +172,5 @@ async def bigquery_projects(self, dataset_id, table_id):
             else:
                 raise Exception(f"Error response from BigQuery: {response}")
         except Exception as e:
-            self.log.exception(f"Error fetching projects")
+            self.log.exception("Error fetching projects")
             return {"error": str(e)}
diff --git a/dataproc_jupyter_plugin/services/clusterListService.py b/dataproc_jupyter_plugin/services/clusterListService.py
deleted file mode 100644
index e83af263..00000000
--- a/dataproc_jupyter_plugin/services/clusterListService.py
+++ /dev/null
@@ -1,46 +0,0 @@
-# Copyright 2023 Google LLC
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# https://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-
-import requests
-from dataproc_jupyter_plugin.commons.constants import CONTENT_TYPE, dataproc_url
-
-
-class ClusterListService:
-    def list_clusters(self, credentials, page_size, page_token, log):
-        try:
-            if (
-                ("access_token" in credentials)
-                and ("project_id" in credentials)
-                and ("region_id" in credentials)
-            ):
-                access_token = credentials["access_token"]
-                project_id = credentials["project_id"]
-                region_id = credentials["region_id"]
-                api_endpoint = f"{dataproc_url}/v1/projects/{project_id}/regions/{region_id}/clusters?pageSize={page_size}&pageToken={page_token}"
-                headers = {
-                    "Content-Type": CONTENT_TYPE,
-                    "Authorization": f"Bearer {access_token}",
-                }
-                response = requests.get(api_endpoint, headers=headers)
-                if response.status_code == 200:
-                    resp = response.json()
-
-                return resp
-            else:
-                log.exception(f"Missing required credentials")
-                raise ValueError("Missing required credentials")
-        except Exception as e:
-            log.exception(f"Error fetching cluster list")
-            return {"error": str(e)}
diff --git a/dataproc_jupyter_plugin/services/composer.py b/dataproc_jupyter_plugin/services/composer.py
new file mode 100644
index 00000000..3700acf8
--- /dev/null
+++ b/dataproc_jupyter_plugin/services/composer.py
@@ -0,0 +1,81 @@
+# Copyright 2024 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+from typing import List
+
+import requests
+
+from dataproc_jupyter_plugin import urls
+from dataproc_jupyter_plugin.commons.constants import (
+    COMPOSER_SERVICE_NAME,
+    CONTENT_TYPE,
+)
+from dataproc_jupyter_plugin.models.models import ComposerEnvironment
+
+
+class Client:
+    def __init__(self, credentials, log):
+        self.log = log
+        if not (
+            ("access_token" in credentials)
+            and ("project_id" in credentials)
+            and ("region_id" in credentials)
+        ):
+            self.log.exception("Missing required credentials")
+            raise ValueError("Missing required credentials")
+        self._access_token = credentials["access_token"]
+        self.project_id = credentials["project_id"]
+        self.region_id = credentials["region_id"]
+
+    def create_headers(self):
+        return {
+            "Content-Type": CONTENT_TYPE,
+            "Authorization": f"Bearer {self._access_token}",
+        }
+
+    async def list_environments(self) -> List[ComposerEnvironment]:
+        try:
+            environments = []
+            composer_url = await urls.gcp_service_url(COMPOSER_SERVICE_NAME)
+            api_endpoint = f"{composer_url}v1/projects/{self.project_id}/locations/{self.region_id}/environments"
+            response = requests.get(api_endpoint, headers=self.create_headers())
+            if response.status_code == 200:
+                resp = response.json()
+                if not resp:
+                    return environments
+                else:
+                    environment = resp["environments"]
+                    # Extract the 'name' values from the 'environments' list
+                    names = [env["name"] for env in environment]
+                    # Extract the last value after the last slash for each 'name'
+                    last_values = [name.split("/")[-1] for name in names]
+                    for env in last_values:
+                        name = env
+                        environments.append(
+                            ComposerEnvironment(
+                                name=name,
+                                label=name,
+                                description=f"Environment: {name}",
+                                file_extensions=["ipynb"],
+                                metadata={"path": env},
+                            )
+                        )
+                    return environments
+            else:
+                self.log.exception("Error listing environments")
+                print(f"Error: {response.status_code} - {response.text}")
+        except FileNotFoundError:
+            environments = []
+            return environments
diff --git a/dataproc_jupyter_plugin/services/composerService.py b/dataproc_jupyter_plugin/services/composerService.py
deleted file mode 100644
index 7c1b20e8..00000000
--- a/dataproc_jupyter_plugin/services/composerService.py
+++ /dev/null
@@ -1,75 +0,0 @@
-# Copyright 2023 Google LLC
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# https://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-
-from typing import Dict, List
-import requests
-from dataproc_jupyter_plugin.models.models import ComposerEnvironment
-from dataproc_jupyter_plugin.commons.constants import CONTENT_TYPE, ENVIRONMENT_API
-
-
-class ComposerService:
-    def list_environments(self, credentials, log) -> List[ComposerEnvironment]:
-        try:
-            if (
-                ("access_token" in credentials)
-                and ("project_id" in credentials)
-                and ("region_id" in credentials)
-            ):
-                access_token = credentials["access_token"]
-                project_id = credentials["project_id"]
-                region_id = credentials["region_id"]
-                environments = []
-                api_endpoint = f"{ENVIRONMENT_API}/projects/{project_id}/locations/{region_id}/environments"
-
-                headers = {
-                    "Content-Type": CONTENT_TYPE,
-                    "Authorization": f"Bearer {access_token}",
-                }
-                response = requests.get(api_endpoint, headers=headers)
-                if response.status_code == 200:
-                    resp = response.json()
-                    if not resp:
-                        return environments
-                    else:
-                        environment = resp["environments"]
-                        # Extract the 'name' values from the 'environments' list
-                        names = [env["name"] for env in environment]
-                        # Extract the last value after the last slash for each 'name'
-                        last_values = [name.split("/")[-1] for name in names]
-                        for env in last_values:
-                            name = env
-                            environments.append(
-                                ComposerEnvironment(
-                                    name=name,
-                                    label=name,
-                                    description=f"Environment: {name}",
-                                    file_extensions=["ipynb"],
-                                    metadata={"path": env},
-                                )
-                            )
-                        return environments
-                else:
-                    log.exception(f"Error listing environments")
-                    print(f"Error: {response.status_code} - {response.text}")
-            else:
-                log.exception(f"Missing required credentials")
-                raise ValueError("Missing required credentials")
-
-        except FileNotFoundError:
-            environments = []
-            return environments
-
-    def manage_environments_command(self) -> str:
-        return ""
diff --git a/dataproc_jupyter_plugin/services/dagListService.py b/dataproc_jupyter_plugin/services/dagListService.py
deleted file mode 100644
index ab90a9d0..00000000
--- a/dataproc_jupyter_plugin/services/dagListService.py
+++ /dev/null
@@ -1,138 +0,0 @@
-# Copyright 2023 Google LLC
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# https://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-
-import subprocess
-import requests
-from dataproc_jupyter_plugin.services.composerService import ENVIRONMENT_API
-from dataproc_jupyter_plugin.commons.constants import CONTENT_TYPE
-
-
-class DagListService:
-    def get_airflow_uri(self, composer_name, credentials, log):
-        try:
-            if (
-                ("access_token" in credentials)
-                and ("project_id" in credentials)
-                and ("region_id" in credentials)
-            ):
-                access_token = credentials["access_token"]
-                project_id = credentials["project_id"]
-                region_id = credentials["region_id"]
-                api_endpoint = f"{ENVIRONMENT_API}/projects/{project_id}/locations/{region_id}/environments/{composer_name}"
-
-                headers = {
-                    "Content-Type": CONTENT_TYPE,
-                    "Authorization": f"Bearer {access_token}",
-                }
-                response = requests.get(api_endpoint, headers=headers)
-                if response.status_code == 200:
-                    resp = response.json()
-                    airflow_uri = resp.get("config", {}).get("airflowUri", "")
-                    bucket = resp.get("storageConfig", {}).get("bucket", "")
-                    return airflow_uri, bucket
-            else:
-                log.exception(f"Missing required credentials")
-                raise ValueError("Missing required credentials")
-        except Exception as e:
-            log.exception(f"Error getting airflow uri: {str(e)}")
-            print(f"Error: {e}")
-
-    def list_jobs(self, credentials, composer_name, tags, log):
-        airflow_uri, bucket = DagListService.get_airflow_uri(
-            self, composer_name, credentials, log
-        )
-        try:
-            if "access_token" in credentials:
-                access_token = credentials["access_token"]
-                api_endpoint = f"{airflow_uri}/api/v1/dags?tags={tags}"
-                headers = {
-                    "Content-Type": CONTENT_TYPE,
-                    "Authorization": f"Bearer {access_token}",
-                }
-                response = requests.get(api_endpoint, headers=headers)
-                if response.status_code == 200:
-                    resp = response.json()
-                return resp, bucket
-            else:
-                log.exception(f"Missing required credentials")
-                raise ValueError("Missing required credentials")
-        except Exception as e:
-            log.exception(f"Error getting dag list: {str(e)}")
-            return {"error": str(e)}
-
-
-class DagDeleteService:
-    def delete_job(self, credentials, composer_name, dag_id, from_page, log):
-        airflow_uri, bucket = DagListService.get_airflow_uri(
-            self, composer_name, credentials, log
-        )
-        try:
-            if "access_token" in credentials:
-                access_token = credentials["access_token"]
-                api_endpoint = f"{airflow_uri}/api/v1/dags/{dag_id}"
-                headers = {
-                    "Content-Type": CONTENT_TYPE,
-                    "Authorization": f"Bearer {access_token}",
-                }
-                if from_page == None:
-                    response = requests.delete(api_endpoint, headers=headers)
-                    log.info(response)
-                cmd = f"gsutil rm gs://{bucket}/dags/dag_{dag_id}.py"
-                process = subprocess.Popen(
-                    cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True
-                )
-                output, _ = process.communicate()
-                if process.returncode == 0:
-                    return 0
-                else:
-                    log.exception(f"Error deleting dag")
-                    return 1
-            else:
-                log.exception(f"Missing required credentials")
-                raise ValueError("Missing required credentials")
-        except Exception as e:
-            log.exception(f"Error deleting dag: {str(e)}")
-            return {"error": str(e)}
-
-
-class DagUpdateService:
-    def update_job(self, credentials, composer_name, dag_id, status, log):
-        airflow_uri, bucket = DagListService.get_airflow_uri(
-            self, composer_name, credentials, log
-        )
-        try:
-            if "access_token" in credentials:
-                access_token = credentials["access_token"]
-                api_endpoint = f"{airflow_uri}/api/v1/dags/{dag_id}"
-                headers = {
-                    "Content-Type": CONTENT_TYPE,
-                    "Authorization": f"Bearer {access_token}",
-                }
-                if status == "true":
-                    data = {"is_paused": False}
-                else:
-                    data = {"is_paused": True}
-                response = requests.patch(api_endpoint, json=data, headers=headers)
-                if response.status_code == 200:
-                    return 0
-                else:
-                    log.exception(f"Error updating status")
-                    return 1
-            else:
-                log.exception(f"Missing required credentials")
-                raise ValueError("Missing required credentials")
-        except Exception as e:
-            log.exception(f"Error updating status: {str(e)}")
-            return {"error": str(e)}
diff --git a/dataproc_jupyter_plugin/services/dagRunService.py b/dataproc_jupyter_plugin/services/dagRunService.py
deleted file mode 100644
index 243fa8ea..00000000
--- a/dataproc_jupyter_plugin/services/dagRunService.py
+++ /dev/null
@@ -1,121 +0,0 @@
-# Copyright 2023 Google LLC
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# https://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-
-from dataproc_jupyter_plugin.services.dagListService import DagListService
-import requests
-from dataproc_jupyter_plugin.services.composerService import ENVIRONMENT_API
-from dataproc_jupyter_plugin.commons.constants import CONTENT_TYPE
-
-
-class DagRunListService:
-    def list_dag_runs(
-        self, credentials, composer_name, dag_id, start_date, end_date, offset, log
-    ):
-        airflow_uri, bucket = DagListService.get_airflow_uri(
-            self, composer_name, credentials, log
-        )
-        try:
-            if (
-                ("access_token" in credentials)
-                and ("project_id" in credentials)
-                and ("region_id" in credentials)
-            ):
-                access_token = credentials["access_token"]
-                api_endpoint = f"{airflow_uri}/api/v1/dags/{dag_id}/dagRuns?execution_date_gte={start_date}&execution_date_lte={end_date}&offset={offset}"
-                headers = {
-                    "Content-Type": CONTENT_TYPE,
-                    "Authorization": f"Bearer {access_token}",
-                }
-                response = requests.get(api_endpoint, headers=headers)
-                if response.status_code == 200:
-                    resp = response.json()
-
-                return resp
-            else:
-                log.exception(f"Missing required credentials")
-                raise ValueError("Missing required credentials")
-        except Exception as e:
-            log.exception(f"Error fetching dag run list: {str(e)}")
-            return {"error": str(e)}
-
-
-class DagRunTaskListService:
-    def list_dag_run_task(self, credentials, composer_name, dag_id, dag_run_id, log):
-        airflow_uri, bucket = DagListService.get_airflow_uri(
-            self, composer_name, credentials, log
-        )
-        try:
-            if (
-                ("access_token" in credentials)
-                and ("project_id" in credentials)
-                and ("region_id" in credentials)
-            ):
-                access_token = credentials["access_token"]
-                api_endpoint = f"{airflow_uri}/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances"
-                headers = {
-                    "Content-Type": CONTENT_TYPE,
-                    "Authorization": f"Bearer {access_token}",
-                }
-                response = requests.get(api_endpoint, headers=headers)
-                if response.status_code == 200:
-                    resp = response.json()
-
-                return resp
-            else:
-                log.exception(f"Missing required credentials")
-                raise ValueError("Missing required credentials")
-        except Exception as e:
-            log.exception(f"Error fetching dag run task list: {str(e)}")
-            return {"error": str(e)}
-
-
-class DagRunTaskLogsListService:
-    def list_dag_run_task_logs(
-        self,
-        credentials,
-        composer_name,
-        dag_id,
-        dag_run_id,
-        task_id,
-        task_try_number,
-        log,
-    ):
-        airflow_uri, bucket = DagListService.get_airflow_uri(
-            self, composer_name, credentials, log
-        )
-        try:
-            if (
-                ("access_token" in credentials)
-                and ("project_id" in credentials)
-                and ("region_id" in credentials)
-            ):
-                access_token = credentials["access_token"]
-                api_endpoint = f"{airflow_uri}/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/logs/{task_try_number}"
-                headers = {
-                    "Content-Type": CONTENT_TYPE,
-                    "Authorization": f"Bearer {access_token}",
-                }
-                response = requests.get(api_endpoint, headers=headers)
-                if response.status_code == 200:
-                    resp = response.text
-                    resp_to_json = {"content": resp}
-
-                return resp_to_json
-            else:
-                log.exception(f"Missing required credentials")
-                raise ValueError("Missing required credentials")
-        except Exception as e:
-            log.exception(f"Error fetching dag run task logs: {str(e)}")
-            return {"error": str(e)}
diff --git a/dataproc_jupyter_plugin/services/dataproc.py b/dataproc_jupyter_plugin/services/dataproc.py
new file mode 100644
index 00000000..f269403d
--- /dev/null
+++ b/dataproc_jupyter_plugin/services/dataproc.py
@@ -0,0 +1,66 @@
+# Copyright 2024 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import requests
+
+from dataproc_jupyter_plugin import urls
+from dataproc_jupyter_plugin.commons.constants import (
+    CONTENT_TYPE,
+    DATAPROC_SERVICE_NAME,
+)
+
+
+class Client:
+    def __init__(self, credentials, log):
+        self.log = log
+        if not (
+            ("access_token" in credentials)
+            and ("project_id" in credentials)
+            and ("region_id" in credentials)
+        ):
+            self.log.exception("Missing required credentials")
+            raise ValueError("Missing required credentials")
+        self._access_token = credentials["access_token"]
+        self.project_id = credentials["project_id"]
+        self.region_id = credentials["region_id"]
+
+    def create_headers(self):
+        return {
+            "Content-Type": CONTENT_TYPE,
+            "Authorization": f"Bearer {self._access_token}",
+        }
+
+    async def list_clusters(self, page_size, page_token):
+        try:
+            dataproc_url = await urls.gcp_service_url(DATAPROC_SERVICE_NAME)
+            api_endpoint = f"{dataproc_url}/v1/projects/{self.project_id}/regions/{self.region_id}/clusters?pageSize={page_size}&pageToken={page_token}"
+            response = requests.get(api_endpoint, headers=self.create_headers())
+            if response.status_code == 200:
+                resp = response.json()
+            return resp
+        except Exception as e:
+            self.log.exception("Error fetching cluster list")
+            return {"error": str(e)}
+
+    async def list_runtime(self, page_size, page_token):
+        try:
+            dataproc_url = await urls.gcp_service_url(DATAPROC_SERVICE_NAME)
+            api_endpoint = f"{dataproc_url}/v1/projects/{self.project_id}/locations/{self.region_id}/sessionTemplates?pageSize={page_size}&pageToken={page_token}"
+            response = requests.get(api_endpoint, headers=self.create_headers())
+            if response.status_code == 200:
+                resp = response.json()
+            return resp
+        except Exception as e:
+            self.log.exception(f"Error fetching runtime list: {str(e)}")
+            return {"error": str(e)}
diff --git a/dataproc_jupyter_plugin/services/downloadOutputService.py b/dataproc_jupyter_plugin/services/downloadOutputService.py
deleted file mode 100644
index d2ec2681..00000000
--- a/dataproc_jupyter_plugin/services/downloadOutputService.py
+++ /dev/null
@@ -1,36 +0,0 @@
-# Copyright 2023 Google LLC
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# https://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-
-import subprocess
-from dataproc_jupyter_plugin.commons.constants import CONTENT_TYPE
-
-
-class DownloadOutputService:
-    def download_dag_output(self, bucket_name, dag_id, dag_run_id, log):
-        try:
-            cmd = f"gsutil cp 'gs://{bucket_name}/dataproc-output/{dag_id}/output-notebooks/{dag_id}_{dag_run_id}.ipynb' ./"
-            process = subprocess.Popen(
-                cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True
-            )
-            output, _ = process.communicate()
-            if process.returncode == 0:
-                return 0
-            else:
-                log.exception(f"Error downloading output notebook file")
-                return 1
-
-        except Exception as e:
-            log.exception(f"Error downloading output notebook file: {str(e)}")
-            return {"error": str(e)}
diff --git a/dataproc_jupyter_plugin/services/editDagService.py b/dataproc_jupyter_plugin/services/editDagService.py
deleted file mode 100644
index a5168b53..00000000
--- a/dataproc_jupyter_plugin/services/editDagService.py
+++ /dev/null
@@ -1,165 +0,0 @@
-# Copyright 2023 Google LLC
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# https://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-
-import os
-import re
-import requests
-import urllib
-from dataproc_jupyter_plugin.commons.constants import CONTENT_TYPE, storage_url
-
-
-class DagEditService:
-    def get_dag_file(self, credentials, dag_id, bucket_name, log):
-        try:
-            if ("access_token" in credentials) and ("project_id" in credentials):
-                access_token = credentials["access_token"]
-                project_id = credentials["project_id"]
-                file_path = f"dags/dag_{dag_id}.py"
-                encoded_path = urllib.parse.quote(file_path, safe="")
-                api_endpoint = (
-                    f"{storage_url}b/{bucket_name}/o/{encoded_path}?alt=media"
-                )
-                headers = {
-                    "Content-Type": CONTENT_TYPE,
-                    "Authorization": f"Bearer {access_token}",
-                    "X-Goog-User-Project": project_id,
-                }
-                response = requests.get(api_endpoint, headers=headers)
-                if response.status_code == 200:
-                    log.info("Dag file response fetched")
-
-                return response.content
-            else:
-                log.exception(f"Missing required credentials")
-                raise ValueError("Missing required credentials")
-        except Exception as e:
-            log.exception(f"Error reading dag file: {str(e)}")
-            return {"error": str(e)}
-
-    def edit_jobs(self, dag_id, bucket_name, credentials, log):
-        try:
-            cluster_name = ""
-            serverless_name = ""
-            email_on_success = "False"
-            stop_cluster_check = "False"
-            mode_selected = "serverless"
-            time_zone = ""
-            pattern = r"parameters\s*=\s*'''(.*?)'''"
-            file_response = DagEditService.get_dag_file(
-                self, credentials, dag_id, bucket_name, log
-            )
-            content_str = file_response.decode("utf-8")
-            file_content = re.sub(r"(?<!\\)\\(?!n)", "", content_str)
-
-            if file_content:
-                for line in file_content.split("\n"):
-                    if "input_notebook" in line:
-                        input_notebook = line.split("=")[-1].strip().strip("'\"")
-                        break
-
-                for line in file_content.split("\n"):
-                    match = re.search(pattern, file_content, re.DOTALL)
-                    if match:
-                        parameters_yaml = match.group(1)
-                        parameters_list = [
-                            line.strip()
-                            for line in parameters_yaml.split("\n")
-                            if line.strip()
-                        ]
-                    else:
-                        parameters_list = []
-
-                for line in file_content.split("\n"):
-                    if "email" in line:
-                        # Extract the email string from the line
-                        email_str = (
-                            line.split(":")[-1].strip().strip("'\"").replace(",", "")
-                        )
-                        # Use regular expression to extract email addresses
-                        email_list = re.findall(
-                            r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b",
-                            email_str,
-                        )
-                        # Remove quotes from email addresses
-                        email_list = [email.strip("'\"") for email in email_list]
-                        break
-                for line in file_content.split("\n"):
-                    if "cluster_name" in line:
-                        cluster_name = (
-                            line.split(":")[-1]
-                            .strip()
-                            .strip("'\"}")
-                            .split("'")[0]
-                            .strip()
-                        )  # Extract project_id from the line
-                    elif "submit_pyspark_job" in line:
-                        mode_selected = "cluster"
-                    elif "'retries'" in line:
-                        retries = line.split(":")[-1].strip().strip("'\"},")
-                        retry_count = int(
-                            retries.strip("'\"")
-                        )  # Extract retry_count from the line
-                    elif "retry_delay" in line:
-                        retry_delay = int(
-                            line.split("int('")[1].split("')")[0]
-                        )  # Extract retry_delay from the line
-                    elif "email_on_failure" in line:
-                        second_part = line.split(":")[1].strip()
-                        email_on_failure = second_part.split("'")[
-                            1
-                        ]  # Extract email_failure from the line
-                    elif "email_on_retry" in line:
-                        second_part = line.split(":")[1].strip()
-                        email_on_retry = second_part.split("'")[1]
-                    elif "email_on_success" in line:
-                        second_part = line.split(":")[1].strip()
-                        email_on_success = second_part.split("'")[1]
-                    elif "schedule_interval" in line:
-                        schedule_interval = (
-                            line.split("=")[-1]
-                            .strip()
-                            .strip("'\"")
-                            .split(",")[0]
-                            .rstrip("'\"")
-                        )  # Extract schedule_interval from the line
-                    elif "stop_cluster_check" in line:
-                        stop_cluster_check = line.split("=")[-1].strip().strip("'\"")
-                    elif "serverless_name" in line:
-                        serverless_name = line.split("=")[-1].strip().strip("'\"")
-                    elif "time_zone" in line:
-                        time_zone = line.split("=")[-1].strip().strip("'\"")
-
-                payload = {
-                    "input_filename": input_notebook,
-                    "parameters": parameters_list,
-                    "mode_selected": mode_selected,
-                    "cluster_name": cluster_name,
-                    "serverless_name": serverless_name,
-                    "retry_count": retry_count,
-                    "retry_delay": retry_delay,
-                    "email_failure": email_on_failure,
-                    "email_delay": email_on_retry,
-                    "email_success": email_on_success,
-                    "email": email_list,
-                    "schedule_value": schedule_interval,
-                    "stop_cluster": stop_cluster_check,
-                    "time_zone": time_zone,
-                }
-                return payload
-
-            else:
-                log.exception(f"No Dag file found")
-        except Exception as e:
-            log.exception(f"Error downloading dag file: {str(e)}")
diff --git a/dataproc_jupyter_plugin/services/executorService.py b/dataproc_jupyter_plugin/services/executor.py
similarity index 56%
rename from dataproc_jupyter_plugin/services/executorService.py
rename to dataproc_jupyter_plugin/services/executor.py
index a65dade6..6cf50b3c 100644
--- a/dataproc_jupyter_plugin/services/executorService.py
+++ b/dataproc_jupyter_plugin/services/executor.py
@@ -1,4 +1,4 @@
-# Copyright 2023 Google LLC
+# Copyright 2024 Google LLC
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -12,110 +12,85 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from datetime import datetime, timedelta
 import os
+import shutil
 import subprocess
-from typing import Dict
 import uuid
+from datetime import datetime, timedelta
 
-from google.cloud.jupyter_config.config import gcp_account
-from jinja2 import Environment, PackageLoader, select_autoescape
-import nbformat
 import pendulum
 import requests
+from google.cloud.jupyter_config.config import gcp_account
+from jinja2 import Environment, PackageLoader, select_autoescape
 
-from dataproc_jupyter_plugin.services.composerService import ENVIRONMENT_API
+from dataproc_jupyter_plugin import urls
+from dataproc_jupyter_plugin.commons.constants import (
+    COMPOSER_SERVICE_NAME,
+    CONTENT_TYPE,
+    GCS,
+    PACKAGE_NAME,
+    WRAPPER_PAPPERMILL_FILE,
+)
 from dataproc_jupyter_plugin.models.models import DescribeJob
-from dataproc_jupyter_plugin.commons.constants import CONTENT_TYPE, GCS
 
 unique_id = str(uuid.uuid4().hex)
 job_id = ""
 job_name = ""
 TEMPLATES_FOLDER_PATH = "dagTemplates"
-ROOT_FOLDER = "dataproc_jupyter_plugin"
+ROOT_FOLDER = PACKAGE_NAME
 
 
-def get_bucket(runtime_env, credentials, log):
-    try:
-        if (
+class Client:
+    def __init__(self, credentials, log):
+        self.log = log
+        if not (
             ("access_token" in credentials)
             and ("project_id" in credentials)
             and ("region_id" in credentials)
         ):
-            access_token = credentials["access_token"]
-            project_id = credentials["project_id"]
-            region_id = credentials["region_id"]
-            api_endpoint = f"{ENVIRONMENT_API}/projects/{project_id}/locations/{region_id}/environments/{runtime_env}"
+            self.log.exception("Missing required credentials")
+            raise ValueError("Missing required credentials")
+        self._access_token = credentials["access_token"]
+        self.project_id = credentials["project_id"]
+        self.region_id = credentials["region_id"]
 
-            headers = {
-                "Content-Type": CONTENT_TYPE,
-                "Authorization": f"Bearer {access_token}",
-            }
+    def create_headers(self):
+        return {
+            "Content-Type": CONTENT_TYPE,
+            "Authorization": f"Bearer {self._access_token}",
+        }
 
-            response = requests.get(api_endpoint, headers=headers)
+    async def get_bucket(self, runtime_env):
+        try:
+            composer_url = await urls.gcp_service_url(COMPOSER_SERVICE_NAME)
+            api_endpoint = f"{composer_url}v1/projects/{self.project_id}/locations/{self.region_id}/environments/{runtime_env}"
+            response = requests.get(api_endpoint, headers=self.create_headers())
             if response.status_code == 200:
                 resp = response.json()
                 gcs_dag_path = resp.get("storageConfig", {}).get("bucket", "")
                 return gcs_dag_path
-        else:
-            log.exception(f"Missing required credentials")
-            raise ValueError("Missing required credentials")
-    except Exception as e:
-        log.exception(f"Error getting bucket name: {str(e)}")
-        print(f"Error: {e}")
-
-
-def check_file_exists(bucket, file_path, log):
-    cmd = f"gsutil ls gs://{bucket}/dataproc-notebooks/{file_path}"
-    process = subprocess.Popen(
-        cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True
-    )
-    output, error = process.communicate()
-    if process.returncode == 0:
-        return True
-    else:
-        if "matched no objects" in error.decode():
-            return False
-        else:
-            log.exception(f"Error cheking file existence: {error.decode()}")
-            raise FileNotFoundError(error.decode)
-
-
-class ExecutorService:
-    """Default execution manager that executes notebooks"""
+        except Exception as e:
+            self.log.exception(f"Error getting bucket name: {str(e)}")
+            print(f"Error: {e}")
 
-    @staticmethod
-    def upload_dag_to_gcs(dag_file, credentials, gcs_dag_bucket, log):
-        if "region_id" in credentials:
-            cmd = f"gsutil cp '{dag_file}' gs://{gcs_dag_bucket}/dags/"
-            process = subprocess.Popen(
-                cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True
-            )
-            output, error = process.communicate()
-        if process.returncode == 0:
-            log.info(f"Dag file uploaded to gcs successfully")
-            os.remove(dag_file)
-        if process.returncode != 0:
-            log.exception(f"Error uploading dag file to gcs: {error.decode()}")
-            raise IOError(error.decode)
-
-    @staticmethod
-    def upload_input_file_to_gcs(input, gcs_dag_bucket, job_name, log):
-        cmd = f"gsutil cp './{input}' gs://{gcs_dag_bucket}/dataproc-notebooks/{job_name}/input_notebooks/"
+    def check_file_exists(self, bucket, file_path):
+        cmd = f"gsutil ls gs://{bucket}/dataproc-notebooks/{file_path}"
         process = subprocess.Popen(
             cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True
         )
         output, error = process.communicate()
         if process.returncode == 0:
-            log.info(f"Input file uploaded to gcs successfully")
+            return True
         else:
-            log.exception(f"Error uploading input file to gcs: {error.decode()}")
-            raise IOError(error.decode)
+            if "matched no objects" in error.decode():
+                return False
+            else:
+                self.log.exception(f"Error cheking file existence: {error.decode()}")
+                raise FileNotFoundError(error.decode)
 
-    @staticmethod
-    def upload_papermill_to_gcs(gcs_dag_bucket, log):
+    def upload_papermill_to_gcs(self, gcs_dag_bucket):
         env = Environment(
-            loader=PackageLoader("dataproc_jupyter_plugin", "dagTemplates"),
+            loader=PackageLoader(PACKAGE_NAME, "dagTemplates"),
             autoescape=select_autoescape(["py"]),
         )
         wrapper_papermill_path = env.get_template("wrapper_papermill.py").filename
@@ -126,23 +101,36 @@ def upload_papermill_to_gcs(gcs_dag_bucket, log):
         output, error = process.communicate()
         print(process.returncode, error, output)
         if process.returncode == 0:
-            log.info(f"Papermill file uploaded to gcs successfully")
+            self.log.info("Papermill file uploaded to gcs successfully")
             print(process.returncode, error, output)
         else:
-            log.exception(f"Error uploading papermill file to gcs: {error.decode()}")
+            self.log.exception(
+                f"Error uploading papermill file to gcs: {error.decode()}"
+            )
+            raise IOError(error.decode)
+
+    def upload_input_file_to_gcs(self, input, gcs_dag_bucket, job_name):
+        cmd = f"gsutil cp './{input}' gs://{gcs_dag_bucket}/dataproc-notebooks/{job_name}/input_notebooks/"
+        process = subprocess.Popen(
+            cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True
+        )
+        output, error = process.communicate()
+        if process.returncode == 0:
+            self.log.info("Input file uploaded to gcs successfully")
+        else:
+            self.log.exception(f"Error uploading input file to gcs: {error.decode()}")
             raise IOError(error.decode)
 
-    @staticmethod
-    def prepare_dag(job, gcs_dag_bucket, dag_file, credentials, log):
-        log.info(f"Generating dag file")
+    def prepare_dag(self, job, gcs_dag_bucket, dag_file):
+        self.log.info("Generating dag file")
         DAG_TEMPLATE_CLUSTER_V1 = "pysparkJobTemplate-v1.txt"
         DAG_TEMPLATE_SERVERLESS_V1 = "pysparkBatchTemplate-v1.txt"
         environment = Environment(
             loader=PackageLoader("dataproc_jupyter_plugin", TEMPLATES_FOLDER_PATH)
         )
-        if ("project_id" in credentials) and ("region_id" in credentials):
-            gcp_project_id = credentials["project_id"]
-            gcp_region_id = credentials["region_id"]
+
+        gcp_project_id = self.project_id
+        gcp_region_id = self.region_id
         user = gcp_account()
         owner = user.split("@")[0]  # getting username from email
         if job.schedule_value == "":
@@ -237,29 +225,75 @@ def prepare_dag(job, gcs_dag_bucket, dag_file, credentials, log):
                 metastore_service=metastore_service,
                 version=version,
             )
-
-        print(content)
-        with open(dag_file, mode="w", encoding="utf-8") as message:
+        LOCAL_DAG_FILE_LOCATION = f"./scheduled-jobs/{job.name}"
+        file_path = os.path.join(LOCAL_DAG_FILE_LOCATION, dag_file)
+        os.makedirs(LOCAL_DAG_FILE_LOCATION, exist_ok=True)
+        with open(file_path, mode="w", encoding="utf-8") as message:
             message.write(content)
+        env = Environment(
+            loader=PackageLoader(PACKAGE_NAME, "dagTemplates"),
+            autoescape=select_autoescape(["py"]),
+        )
+        wrapper_papermill_path = env.get_template("wrapper_papermill.py").filename
+        shutil.copy2(wrapper_papermill_path, LOCAL_DAG_FILE_LOCATION)
+
+    def upload_dag_to_gcs(self, job, dag_file, gcs_dag_bucket):
+        LOCAL_DAG_FILE_LOCATION = f"./scheduled-jobs/{job.name}"
+        file_path = os.path.join(LOCAL_DAG_FILE_LOCATION, dag_file)
+        cmd = f"gsutil cp '{file_path}' gs://{gcs_dag_bucket}/dags/"
+        process = subprocess.Popen(
+            cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True
+        )
+        output, error = process.communicate()
+        if process.returncode == 0:
+            self.log.info("Dag file uploaded to gcs successfully")
 
-    def execute(self, credentials, input_data, log):
-        job = DescribeJob(**input_data)
-        global job_id
-        global job_name
-        job_id = job.dag_id
-        job_name = job.name
-        dag_file = f"dag_{job_name}.py"
-        gcs_dag_bucket = get_bucket(job.composer_environment_name, credentials, log)
-        remote_file_path = "wrapper_papermill.py"
+        if process.returncode != 0:
+            self.log.exception(f"Error uploading dag file to gcs: {error.decode()}")
+            raise IOError(error.decode)
 
-        if check_file_exists(gcs_dag_bucket, remote_file_path, log):
-            print(f"The file gs://{gcs_dag_bucket}/{remote_file_path} exists.")
-        else:
-            self.upload_papermill_to_gcs(gcs_dag_bucket, log)
-            print(f"The file gs://{gcs_dag_bucket}/{remote_file_path} does not exist.")
-        if not job.input_filename.startswith(GCS):
-            self.upload_input_file_to_gcs(
-                job.input_filename, gcs_dag_bucket, job_name, log
+    async def execute(self, input_data):
+        try:
+            job = DescribeJob(**input_data)
+            global job_id
+            global job_name
+            job_id = job.dag_id
+            job_name = job.name
+            dag_file = f"dag_{job_name}.py"
+            gcs_dag_bucket = await self.get_bucket(job.composer_environment_name)
+            wrapper_pappermill_file_path = WRAPPER_PAPPERMILL_FILE
+
+            if self.check_file_exists(gcs_dag_bucket, wrapper_pappermill_file_path):
+                print(
+                    f"The file gs://{gcs_dag_bucket}/{wrapper_pappermill_file_path} exists."
+                )
+            else:
+                self.upload_papermill_to_gcs(gcs_dag_bucket)
+                print(
+                    f"The file gs://{gcs_dag_bucket}/{wrapper_pappermill_file_path} does not exist."
+                )
+            if not job.input_filename.startswith(GCS):
+                self.upload_input_file_to_gcs(
+                    job.input_filename, gcs_dag_bucket, job_name
+                )
+            self.prepare_dag(job, gcs_dag_bucket, dag_file)
+            self.upload_dag_to_gcs(job, dag_file, gcs_dag_bucket)
+            return {"status": 0}
+        except Exception as e:
+            return {"error": str(e)}
+
+    def download_dag_output(self, bucket_name, dag_id, dag_run_id):
+        try:
+            cmd = f"gsutil cp 'gs://{bucket_name}/dataproc-output/{dag_id}/output-notebooks/{dag_id}_{dag_run_id}.ipynb' ./"
+            process = subprocess.Popen(
+                cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True
             )
-        self.prepare_dag(job, gcs_dag_bucket, dag_file, credentials, log)
-        self.upload_dag_to_gcs(dag_file, credentials, gcs_dag_bucket, log)
+            output, _ = process.communicate()
+            if process.returncode == 0:
+                return 0
+            else:
+                self.log.exception("Error downloading output notebook file")
+                return 1
+        except Exception as e:
+            self.log.exception(f"Error downloading output notebook file: {str(e)}")
+            return {"error": str(e)}
diff --git a/dataproc_jupyter_plugin/services/gcpUrlService.py b/dataproc_jupyter_plugin/services/gcpUrlService.py
deleted file mode 100644
index 4bd82cd6..00000000
--- a/dataproc_jupyter_plugin/services/gcpUrlService.py
+++ /dev/null
@@ -1,25 +0,0 @@
-# Copyright 2023 Google LLC
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# https://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-
-from google.cloud.jupyter_config.config import get_gcloud_config
-
-
-def gcp_service_url(service_name, default_url=None):
-    default_url = default_url or f"https://{service_name}.googleapis.com/"
-    configured_url = get_gcloud_config(
-        f"configuration.properties.api_endpoint_overrides.{service_name}"
-    )
-    url = configured_url or default_url
-    return url
diff --git a/dataproc_jupyter_plugin/services/importErrorService.py b/dataproc_jupyter_plugin/services/importErrorService.py
deleted file mode 100644
index 48dadba2..00000000
--- a/dataproc_jupyter_plugin/services/importErrorService.py
+++ /dev/null
@@ -1,50 +0,0 @@
-# Copyright 2023 Google LLC
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# https://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-
-import requests
-from dataproc_jupyter_plugin.services.dagListService import DagListService
-from dataproc_jupyter_plugin.commons.constants import CONTENT_TYPE
-
-
-class ImportErrorService:
-    def list_import_errors(self, credentials, composer, log):
-        airflow_uri, bucket = DagListService.get_airflow_uri(
-            self, composer, credentials, log
-        )
-        try:
-            if (
-                ("access_token" in credentials)
-                and ("project_id" in credentials)
-                and ("region_id" in credentials)
-            ):
-                access_token = credentials["access_token"]
-                api_endpoint = (
-                    f"{airflow_uri}/api/v1/importErrors?order_by=-import_error_id"
-                )
-
-                headers = {
-                    "Content-Type": CONTENT_TYPE,
-                    "Authorization": f"Bearer {access_token}",
-                }
-                response = requests.get(api_endpoint, headers=headers)
-                if response.status_code == 200:
-                    resp = response.json()
-                return resp
-            else:
-                log.exception(f"Missing required credentials")
-                raise ValueError("Missing required credentials")
-        except Exception as e:
-            log.exception(f"Error fetching import error list: {str(e)}")
-            return {"error": str(e)}
diff --git a/dataproc_jupyter_plugin/services/runtimeListService.py b/dataproc_jupyter_plugin/services/runtimeListService.py
deleted file mode 100644
index 25391d5c..00000000
--- a/dataproc_jupyter_plugin/services/runtimeListService.py
+++ /dev/null
@@ -1,47 +0,0 @@
-# Copyright 2023 Google LLC
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# https://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-
-import requests
-from dataproc_jupyter_plugin.commons.constants import CONTENT_TYPE, dataproc_url
-
-
-class RuntimeListService:
-    def list_runtime(self, credentials, page_size, page_token, log):
-        try:
-            if (
-                ("access_token" in credentials)
-                and ("project_id" in credentials)
-                and ("region_id" in credentials)
-            ):
-                access_token = credentials["access_token"]
-                project_id = credentials["project_id"]
-                region_id = credentials["region_id"]
-                api_endpoint = f"{dataproc_url}/v1/projects/{project_id}/locations/{region_id}/sessionTemplates?pageSize={page_size}&pageToken={page_token}"
-
-                headers = {
-                    "Content-Type": CONTENT_TYPE,
-                    "Authorization": f"Bearer {access_token}",
-                }
-                response = requests.get(api_endpoint, headers=headers)
-                if response.status_code == 200:
-                    resp = response.json()
-
-                return resp
-            else:
-                log.exception(f"Missing required credentials")
-                raise ValueError("Missing required credentials")
-        except Exception as e:
-            log.exception(f"Error fetching runtime list: {str(e)}")
-            return {"error": str(e)}
diff --git a/dataproc_jupyter_plugin/services/triggerDagService.py b/dataproc_jupyter_plugin/services/triggerDagService.py
deleted file mode 100644
index 82c46ec2..00000000
--- a/dataproc_jupyter_plugin/services/triggerDagService.py
+++ /dev/null
@@ -1,50 +0,0 @@
-# Copyright 2023 Google LLC
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# https://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-
-import requests
-from dataproc_jupyter_plugin.services.dagListService import DagListService
-from dataproc_jupyter_plugin.commons.constants import CONTENT_TYPE
-
-
-class TriggerDagService:
-    def dag_trigger(self, credentials, dag_id, composer, log):
-        airflow_uri, bucket = DagListService.get_airflow_uri(
-            self, composer, credentials, log
-        )
-        try:
-            if (
-                ("access_token" in credentials)
-                and ("project_id" in credentials)
-                and ("region_id" in credentials)
-            ):
-                access_token = credentials["access_token"]
-                api_endpoint = f"{airflow_uri}/api/v1/dags/{dag_id}/dagRuns"
-
-                headers = {
-                    "Content-Type": CONTENT_TYPE,
-                    "Authorization": f"Bearer {access_token}",
-                }
-                body = {"conf": {}}
-                response = requests.post(api_endpoint, headers=headers, json=body)
-                if response.status_code == 200:
-                    resp = response.json()
-
-                return resp
-            else:
-                log.exception(f"Missing required credentials")
-                raise ValueError("Missing required credentials")
-        except Exception as e:
-            log.exception(f"Error triggering dag: {str(e)}")
-            return {"error": str(e)}
diff --git a/dataproc_jupyter_plugin/urls.py b/dataproc_jupyter_plugin/urls.py
index 5e2602bf..544a52a6 100644
--- a/dataproc_jupyter_plugin/urls.py
+++ b/dataproc_jupyter_plugin/urls.py
@@ -11,21 +11,32 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-
 from google.cloud import jupyter_config
 
+from dataproc_jupyter_plugin.commons.constants import (
+    CLOUDKMS_SERVICE_NAME,
+    CLOUDRESOURCEMANAGER_SERVICE_NAME,
+    COMPUTE_SERVICE_DEFAULT_URL,
+    COMPUTE_SERVICE_NAME,
+    DATACATALOG_SERVICE_NAME,
+    DATAPROC_SERVICE_NAME,
+    METASTORE_SERVICE_NAME,
+    STORAGE_SERVICE_DEFAULT_URL,
+    STORAGE_SERVICE_NAME,
+)
+
 
 async def map():
-    dataproc_url = await gcp_service_url("dataproc")
+    dataproc_url = await gcp_service_url(DATAPROC_SERVICE_NAME)
     compute_url = await gcp_service_url(
-        "compute", default_url="https://compute.googleapis.com/compute/v1"
+        COMPUTE_SERVICE_NAME, default_url=COMPUTE_SERVICE_DEFAULT_URL
     )
-    metastore_url = await gcp_service_url("metastore")
-    cloudkms_url = await gcp_service_url("cloudkms")
-    cloudresourcemanager_url = await gcp_service_url("cloudresourcemanager")
-    datacatalog_url = await gcp_service_url("datacatalog")
+    metastore_url = await gcp_service_url(METASTORE_SERVICE_NAME)
+    cloudkms_url = await gcp_service_url(CLOUDKMS_SERVICE_NAME)
+    cloudresourcemanager_url = await gcp_service_url(CLOUDRESOURCEMANAGER_SERVICE_NAME)
+    datacatalog_url = await gcp_service_url(DATACATALOG_SERVICE_NAME)
     storage_url = await gcp_service_url(
-        "storage", default_url="https://storage.googleapis.com/storage/v1/"
+        STORAGE_SERVICE_NAME, default_url=STORAGE_SERVICE_DEFAULT_URL
     )
     url_map = {
         "dataproc_url": dataproc_url,
diff --git a/package.json b/package.json
index 845584d0..86558b53 100644
--- a/package.json
+++ b/package.json
@@ -1,6 +1,6 @@
 {
     "name": "dataproc_jupyter_plugin",
-    "version": "0.1.77",
+    "version": "0.1.78",
     "description": "It is a plugin to work with dataproc services in Jupyterlab",
     "keywords": [
         "jupyter",
diff --git a/src/batches/batchService.tsx b/src/batches/batchService.tsx
index d26b9862..c1c99e9c 100644
--- a/src/batches/batchService.tsx
+++ b/src/batches/batchService.tsx
@@ -290,7 +290,7 @@ export class BatchService {
       setRegionName(credentials.region_id || '');
       setProjectName(credentials.project_id || '');
       loggedFetch(
-        `${DATAPROC}/projects/${credentials.project_id}/locations/${credentials.region_id}/batches?orderBy=create_time desc&&pageSize=50&pageToken=${pageToken}`,
+        `${DATAPROC}/projects/${credentials.project_id}/locations/${credentials.region_id}/batches?orderBy=create_time desc&&pageSize=500&pageToken=${pageToken}`,
         {
           headers: {
             'Content-Type': API_HEADER_CONTENT_TYPE,
@@ -335,10 +335,12 @@ export class BatchService {
                 );
               }
               if (responseResult?.error?.code) {
-                toast.error(
-                  responseResult?.error?.message,
-                  toastifyCustomStyle
-                );
+                if (!toast.isActive('batchListingError')) {
+                  toast.error(responseResult?.error?.message, {
+                    ...toastifyCustomStyle,
+                    toastId: 'batchListingError'
+                  });
+                }
               }
               const existingBatchData = previousBatchesList ?? [];
 
@@ -509,7 +511,6 @@ export class BatchService {
               transformedNetworkSelected = responseResult.network.split('/')[9];
 
               setNetworkSelected(transformedNetworkSelected);
-              setIsloadingNetwork(false);
               if (responseResult?.error?.code) {
                 toast.error(
                   responseResult?.error?.message,
@@ -522,7 +523,6 @@ export class BatchService {
             });
         })
         .catch((err: Error) => {
-          setIsloadingNetwork(false);
           DataprocLoggingService.log(
             'Error selecting Network',
             LOG_LEVEL.ERROR
@@ -534,8 +534,11 @@ export class BatchService {
 
   static listNetworksAPIService = async (
     setNetworklist: (value: string[]) => void,
-    setNetworkSelected: (value: string) => void
+    setNetworkSelected: (value: string) => void,
+    batchInfoResponse: any,
+    setIsloadingNetwork: (value: boolean) => void
   ) => {
+    setIsloadingNetwork(true);
     const credentials = await authApi();
     const { COMPUTE } = await gcpServiceUrls;
     if (credentials) {
@@ -571,7 +574,9 @@ export class BatchService {
                   }
                 );
                 setNetworklist(transformedNetworkList);
-                setNetworkSelected(transformedNetworkList[0]);
+                if (batchInfoResponse === undefined) {
+                  setNetworkSelected(transformedNetworkList[0]);
+                }
                 if (responseResult?.error?.code) {
                   toast.error(
                     responseResult?.error?.message,
@@ -699,8 +704,11 @@ export class BatchService {
   static listSubNetworksAPIService = async (
     subnetwork: string,
     setSubNetworklist: (value: string[]) => void,
-    setSubNetworkSelected: (value: string) => void
+    setSubNetworkSelected: (value: string) => void,
+    batchInfoResponse: any,
+    setIsloadingNetwork: (value: boolean) => void
   ) => {
+    setIsloadingNetwork(true);
     const credentials = await authApi();
     const { COMPUTE } = await gcpServiceUrls;
     if (credentials) {
@@ -737,7 +745,10 @@ export class BatchService {
                   (data: { name: string }) => data.name
                 );
                 setSubNetworklist(transformedServiceList);
-                setSubNetworkSelected(transformedServiceList[0]);
+                if (batchInfoResponse === undefined) {
+                  setSubNetworkSelected(transformedServiceList[0]);
+                }
+                setIsloadingNetwork(false);
                 if (responseResult?.error?.code) {
                   toast.error(
                     responseResult?.error?.message,
@@ -755,6 +766,7 @@ export class BatchService {
             'Error listing subNetworks',
             LOG_LEVEL.ERROR
           );
+          setIsloadingNetwork(false);
           toast.error(
             `Error listing subNetworks : ${err}`,
             toastifyCustomStyle
diff --git a/src/batches/createBatch.tsx b/src/batches/createBatch.tsx
index 0067b403..6c927d28 100644
--- a/src/batches/createBatch.tsx
+++ b/src/batches/createBatch.tsx
@@ -42,7 +42,6 @@ import LabelProperties from '../jobs/labelProperties';
 import { authApi } from '../utils/utils';
 import ErrorPopup from '../utils/errorPopup';
 import errorIcon from '../../style/icons/error_icon.svg';
-import { Select } from '../controls/MuiWrappedSelect';
 import { Input } from '../controls/MuiWrappedInput';
 import {
   Autocomplete,
@@ -76,16 +75,16 @@ interface ICreateBatchProps {
 }
 
 function batchTypeFunction(batchKey: string) {
-  let batchType = 'spark';
+  let batchType = 'Spark';
   switch (batchKey) {
     case 'sparkRBatch':
-      batchType = 'sparkR';
+      batchType = 'SparkR';
       return batchType;
     case 'pysparkBatch':
-      batchType = 'pySpark';
+      batchType = 'PySpark';
       return batchType;
     case 'sparkSqlBatch':
-      batchType = 'sparkSql';
+      batchType = 'SparkSql';
       return batchType;
     default:
       return batchType;
@@ -101,7 +100,7 @@ function CreateBatch({
   createBatch
 }: ICreateBatchProps) {
   let batchKeys: string[] = [];
-  let batchType = 'spark';
+  let batchType = 'Spark';
   let mainClass = '';
   let mainJarFileUri = '';
   let mainRFileUri = '';
@@ -170,8 +169,8 @@ function CreateBatch({
       keyType =
         batchInfoResponse?.environmentConfig?.executionConfig?.kmsKey || '';
       const keyringValues = keyType.split('/'); // splitting keyrings and key form projects/projectName/locations/regionName/keyRings/keyRing/cryptoKeys/key
-      keyRing = keyringValues[5];
-      keys = keyringValues[7];
+      keyRing = keyringValues[5] ? keyringValues[5] : '';
+      keys = keyringValues[7] ? keyringValues[7] : '';
       historyServerValue =
         batchInfoResponse?.environmentConfig?.peripheralsConfig
           ?.sparkHistoryServerConfig?.dataprocCluster || '';
@@ -287,7 +286,6 @@ function CreateBatch({
   );
   const [sharedvpcSelected, setSharedvpcSelected] = useState('');
   const [projectInfo, setProjectInfo] = useState('');
-  const [labelFocused, setLabelFocused] = useState(false);
   const handleCreateBatchBackView = () => {
     if (setCreateBatchView) {
       setCreateBatchView(false);
@@ -315,19 +313,25 @@ function CreateBatch({
     setKeySelected('');
   };
   useEffect(() => {
-    listKeysAPI(keyRingSelected);
+    if (keyRingSelected !== '') {
+      listKeysAPI(keyRingSelected);
+    }
   }, [keyRingSelected]);
+
   useEffect(() => {
-    listSubNetworksAPI(networkSelected);
+    if (
+      batchInfoResponse === undefined ||
+      batchInfoResponse?.environmentConfig?.executionConfig?.subnetworkUri
+    ) {
+      if (networkSelected !== '') {
+        listSubNetworksAPI(networkSelected);
+      }
+    } else {
+      listSubNetworksAPI(networkSelected);
+    }
   }, [networkSelected]);
   useEffect(() => {
-    const batchTypeData = [
-      { key: 'spark', value: 'spark', text: 'Spark' },
-      { key: 'sparkR', value: 'sparkR', text: 'SparkR' },
-      { key: 'sparkSql', value: 'sparkSql', text: 'SparkSql' },
-      { key: 'pySpark', value: 'pySpark', text: 'PySpark' }
-    ];
-
+    const batchTypeData = ['Spark', 'SparkR', 'SparkSql', 'PySpark'];
     setBatchTypeList(batchTypeData);
 
     listClustersAPI();
@@ -495,7 +499,7 @@ function CreateBatch({
         networkList.length !== 0 &&
         subNetworkList.length === 0);
     switch (batchTypeSelected) {
-      case 'spark':
+      case 'Spark':
         return (
           commonConditions ||
           (selectedRadio === 'mainClass' && mainClassSelected === '') ||
@@ -511,7 +515,7 @@ function CreateBatch({
           networkTagsDuplicateValidation ||
           jarFileDuplicateValidation
         );
-      case 'sparkR':
+      case 'SparkR':
         return (
           commonConditions ||
           mainRSelected === '' ||
@@ -524,7 +528,7 @@ function CreateBatch({
           argumentsDuplicateValidation ||
           networkTagsDuplicateValidation
         );
-      case 'pySpark':
+      case 'PySpark':
         return (
           commonConditions ||
           mainPythonSelected === '' ||
@@ -541,7 +545,7 @@ function CreateBatch({
           networkTagsDuplicateValidation ||
           jarFileDuplicateValidation
         );
-      case 'sparkSql':
+      case 'SparkSql':
         return (
           commonConditions ||
           queryFileSelected === '' ||
@@ -635,7 +639,9 @@ function CreateBatch({
   const listNetworksAPI = async () => {
     await BatchService.listNetworksAPIService(
       setNetworklist,
-      setNetworkSelected
+      setNetworkSelected,
+      batchInfoResponse,
+      setIsloadingNetwork
     );
   };
 
@@ -651,7 +657,9 @@ function CreateBatch({
     await BatchService.listSubNetworksAPIService(
       subnetwork,
       setSubNetworklist,
-      setSubNetworkSelected
+      setSubNetworkSelected,
+      batchInfoResponse,
+      setIsloadingNetwork
     );
   };
 
@@ -669,8 +677,10 @@ function CreateBatch({
   };
 
   const handleSharedSubNetwork = async (data: string | null) => {
-    setSharedvpcSelected(data!.toString());
-    await handleProjectIdChange(projectId, data!.toString());
+    if (data !== null) {
+      setSharedvpcSelected(data!.toString());
+      await handleProjectIdChange(projectId, data!.toString());
+    }
   };
 
   type Payload = {
@@ -716,7 +726,7 @@ function CreateBatch({
   ): Payload => {
     const payload: Payload = {};
 
-    if (batchTypeSelected === 'spark') {
+    if (batchTypeSelected === 'Spark') {
       payload.sparkBatch = {
         ...(mainJarSelected !== '' && { mainJarFileUri: mainJarSelected }),
         ...(mainClassSelected !== '' && { mainClass: mainClassSelected }),
@@ -730,7 +740,7 @@ function CreateBatch({
         ...(argumentsSelected.length > 0 && { args: argumentsSelected })
       };
     }
-    if (batchTypeSelected === 'sparkR') {
+    if (batchTypeSelected === 'SparkR') {
       payload.sparkRBatch = {
         ...(mainRSelected !== '' && { mainRFileUri: mainRSelected }),
         ...(ArchiveFilesSelected.length > 0 && {
@@ -742,7 +752,7 @@ function CreateBatch({
         ...(argumentsSelected.length > 0 && { args: argumentsSelected })
       };
     }
-    if (batchTypeSelected === 'pySpark') {
+    if (batchTypeSelected === 'PySpark') {
       payload.pysparkBatch = {
         ...(additionalPythonFileSelected.length > 0 && {
           pythonFileUris: additionalPythonFileSelected
@@ -760,7 +770,7 @@ function CreateBatch({
         ...(argumentsSelected.length > 0 && { args: argumentsSelected })
       };
     }
-    if (batchTypeSelected === 'sparkSql') {
+    if (batchTypeSelected === 'SparkSql') {
       payload.sparkSqlBatch = {
         ...(queryFileSelected !== '' && { queryFileUri: queryFileSelected }),
         ...(parameterObject && { queryVariables: parameterObject }),
@@ -916,11 +926,10 @@ function CreateBatch({
     setSharedvpcSelected('');
   };
 
-  const handleBatchTypeSelected = (
-    event: React.SyntheticEvent<HTMLElement, Event>,
-    data: DropdownProps
-  ) => {
-    setBatchTypeSelected(data.value!.toString());
+  const handleBatchTypeSelected = (data: DropdownProps | null) => {
+    if (data !== null) {
+      setBatchTypeSelected(data!.toString());
+    }
     setFilesSelected([]);
     setJarFilesSelected([]);
     setAdditionalPythonFileSelected([]);
@@ -933,7 +942,9 @@ function CreateBatch({
   };
 
   const handleServiceSelected = (data: string | null) => {
-    setServicesSelected(data!.toString());
+    if (data !== null) {
+      setServicesSelected(data!.toString());
+    }
   };
 
   const handleProjectIdChange = (data: any, network: string | undefined) => {
@@ -943,19 +954,27 @@ function CreateBatch({
     regionListAPI(data, network);
   };
   const handleNetworkChange = async (data: DropdownProps | null) => {
-    setNetworkSelected(data!.toString());
-    await listSubNetworksAPI(data!.toString());
-    await handleProjectIdChange(projectId, data!.toString());
+    if (data !== null) {
+      setNetworkSelected(data!.toString());
+      await listSubNetworksAPI(data!.toString());
+      await handleProjectIdChange(projectId, data!.toString());
+    }
   };
   const handleSubNetworkChange = (data: string | null) => {
-    setSubNetworkSelected(data!.toString());
+    if (data !== null) {
+      setSubNetworkSelected(data!.toString());
+    }
   };
   const handleKeyRingChange = (data: string | null) => {
-    setKeyRingSelected(data!.toString());
-    listKeysAPI(data!.toString());
+    if (data !== null) {
+      setKeyRingSelected(data!.toString());
+      listKeysAPI(data!.toString());
+    }
   };
   const handlekeyChange = (data: string | null) => {
-    setKeySelected(data!.toString());
+    if (data !== null) {
+      setKeySelected(data!.toString());
+    }
   };
   const handleMainClassSelected = (value: string) => {
     setMainClassUpdated(true);
@@ -982,12 +1001,6 @@ function CreateBatch({
 
     setManualKeySelected(inputValue);
   };
-  const handleSelectFocus = () => {
-    setLabelFocused(true);
-  };
-  const handleBlur = () => {
-    setLabelFocused(false);
-  };
   return (
     <div>
       <div className="cluster-details-header">
@@ -1031,25 +1044,14 @@ function CreateBatch({
           </div>
           <div className="submit-job-label-header">Container</div>
           <div className="select-text-overlay">
-            <label
-              className={
-                labelFocused
-                  ? 'select-dropdown-text label-focused'
-                  : 'select-dropdown-text'
-              }
-              htmlFor="batch-type"
-            >
-              Batch type*
-            </label>
-            <Select
-              search
+            <Autocomplete
               className="project-region-select"
               value={batchTypeSelected}
-              type="text"
               options={batchTypeList}
-              onChange={handleBatchTypeSelected}
-              onFocus={handleSelectFocus}
-              onBlur={handleBlur}
+              onChange={(_event, val) => handleBatchTypeSelected(val)}
+              renderInput={params => (
+                <TextField {...params} label=" Batch type*" />
+              )}
             />
           </div>
           <div className="select-text-overlay">
@@ -1061,7 +1063,7 @@ function CreateBatch({
               Label="Runtime version*"
             />
           </div>
-          {batchTypeSelected === 'spark' && (
+          {batchTypeSelected === 'Spark' && (
             <div>
               <div>
                 <div className="create-batch-radio">
@@ -1108,7 +1110,7 @@ function CreateBatch({
               )}
             </div>
           )}
-          {batchTypeSelected === 'spark' && (
+          {batchTypeSelected === 'Spark' && (
             <div>
               <div>
                 <div className="create-batch-radio">
@@ -1167,7 +1169,7 @@ function CreateBatch({
               )}
             </div>
           )}
-          {batchTypeSelected === 'sparkR' && (
+          {batchTypeSelected === 'SparkR' && (
             <>
               <div className="select-text-overlay">
                 <Input
@@ -1201,7 +1203,7 @@ function CreateBatch({
               )}
             </>
           )}
-          {batchTypeSelected === 'pySpark' && (
+          {batchTypeSelected === 'PySpark' && (
             <>
               <div className="select-text-overlay">
                 <Input
@@ -1235,7 +1237,7 @@ function CreateBatch({
               )}
             </>
           )}
-          {batchTypeSelected === 'pySpark' && (
+          {batchTypeSelected === 'PySpark' && (
             <>
               <div className="select-text-overlay">
                 <MuiChipsInput
@@ -1273,7 +1275,7 @@ function CreateBatch({
               )}
             </>
           )}
-          {batchTypeSelected === 'sparkSql' && (
+          {batchTypeSelected === 'SparkSql' && (
             <>
               <div className="select-text-overlay">
                 <Input
@@ -1352,7 +1354,7 @@ function CreateBatch({
             </div>
           </div>
           {
-            batchTypeSelected !== 'sparkR' && (
+            batchTypeSelected !== 'SparkR' && (
               <>
                 <div className="select-text-overlay">
                   <MuiChipsInput
@@ -1402,7 +1404,7 @@ function CreateBatch({
             )
             //) )
           }
-          {batchTypeSelected !== 'sparkSql' && (
+          {batchTypeSelected !== 'SparkSql' && (
             <>
               <div className="select-text-overlay">
                 <MuiChipsInput
@@ -1443,7 +1445,7 @@ function CreateBatch({
               )}
             </>
           )}
-          {batchTypeSelected !== 'sparkSql' && (
+          {batchTypeSelected !== 'SparkSql' && (
             <>
               <div className="select-text-overlay">
                 <MuiChipsInput
@@ -1486,7 +1488,7 @@ function CreateBatch({
               )}
             </>
           )}
-          {batchTypeSelected !== 'sparkSql' && (
+          {batchTypeSelected !== 'SparkSql' && (
             <>
               <div className="select-text-overlay">
                 <MuiChipsInput
@@ -1513,7 +1515,7 @@ function CreateBatch({
               )}
             </>
           )}
-          {batchTypeSelected === 'sparkSql' && (
+          {batchTypeSelected === 'SparkSql' && (
             <>
               <div className="submit-job-label-header">Query parameters</div>
               <LabelProperties
@@ -1643,7 +1645,8 @@ function CreateBatch({
                   </div>
                 </div>
               )}
-            {selectedNetworkRadio === 'projectNetwork' &&
+            {!isloadingNetwork &&
+              selectedNetworkRadio === 'projectNetwork' &&
               networkList.length !== 0 &&
               subNetworkList.length === 0 && (
                 <div className="error-key-parent">
diff --git a/src/bigQuery/bigQueryService.tsx b/src/bigQuery/bigQueryService.tsx
index 31a2aee3..3cd3ff53 100644
--- a/src/bigQuery/bigQueryService.tsx
+++ b/src/bigQuery/bigQueryService.tsx
@@ -39,6 +39,7 @@ export class BigQueryService {
     setTotalRowSize: (value: string) => void,
     setPreviewDataList: any
   ) => {
+    setIsLoading(true);
     try {
       const startIndex = pageIndex * maxResults;
       const data: any = await requestAPI(
diff --git a/src/cluster/clusterServices.tsx b/src/cluster/clusterServices.tsx
index 421753bf..0a8d6252 100644
--- a/src/cluster/clusterServices.tsx
+++ b/src/cluster/clusterServices.tsx
@@ -140,7 +140,12 @@ export class ClusterService {
         setLoggedIn(true);
       }
       if (formattedResponse?.error?.code) {
-        toast.error(formattedResponse?.error?.message, toastifyCustomStyle);
+        if (!toast.isActive('clusterListingError')) {
+          toast.error(formattedResponse?.error?.message, {
+            ...toastifyCustomStyle,
+            toastId: 'clusterListingError'
+          });
+        }
       }
     } catch (error) {
       setIsLoading(false);
diff --git a/src/controls/DynamicDropdown.tsx b/src/controls/DynamicDropdown.tsx
index 2d85ba5b..d60cbd11 100644
--- a/src/controls/DynamicDropdown.tsx
+++ b/src/controls/DynamicDropdown.tsx
@@ -15,10 +15,17 @@
  * limitations under the License.
  */
 
-import React, { useEffect, useMemo, useRef, useState } from 'react';
+import React, {
+  useEffect,
+  useMemo,
+  useRef,
+  useState,
+  useCallback
+} from 'react';
 import TextField from '@mui/material/TextField';
 import Autocomplete, { AutocompleteProps } from '@mui/material/Autocomplete';
 import { ChipTypeMap, Paper, PaperProps } from '@mui/material';
+import { handleDebounce } from '../utils/utils';
 
 type Props = {
   fetchFunc: (search: string) => Promise<string[]>;
@@ -26,7 +33,7 @@ type Props = {
 };
 
 /**
- * Component to render a dyanmic selector dropdown.
+ * Component to render a dynamic selector dropdown.
  */
 export function DynamicDropdown(
   props: Props &
@@ -45,22 +52,29 @@ export function DynamicDropdown(
   const [search, setSearch] = useState('');
   const [filteredList, setFilteredList] = useState<string[]>([]);
   const currentSearch = useRef(search);
+
+  const debouncedFetch = useCallback(
+    handleDebounce((search: string) => {
+      currentSearch.current = search;
+      fetchFunc(search).then(items => {
+        if (currentSearch.current !== search) {
+          //To update the results as per change in prefix
+          return;
+        }
+        setFilteredList(items);
+      });
+    }, 500), // 500ms debounce
+    [fetchFunc]
+  );
+
   useEffect(() => {
-    currentSearch.current = search;
-    fetchFunc(search).then(items => {
-      if (currentSearch.current != search) {
-        // The prefix changed while the network request was pending
-        // so we should throw away these results.
-        return;
-      }
-      setFilteredList(items);
-    });
-  }, [search, fetchFunc]);
+    debouncedFetch(search);
+  }, [search, debouncedFetch]);
 
   /**
-   * This is the last selected value when the dropdown is opened.  We
+   * This is the last selected value when the dropdown is opened. We
    * always ensure that the current ID exists in the dropdown list,
-   * preppending it if necessary.
+   * prepending it if necessary.
    */
   const [hoistedValue, setHoistedValue] = useState(value);
 
@@ -76,6 +90,7 @@ export function DynamicDropdown(
     }
     return filteredList;
   }, [filteredList, hoistedValue]) as string[];
+
   return (
     <Autocomplete
       value={value}
diff --git a/src/dpms/dpmsService.tsx b/src/dpms/dpmsService.tsx
index c310bed4..2fd495f8 100644
--- a/src/dpms/dpmsService.tsx
+++ b/src/dpms/dpmsService.tsx
@@ -167,24 +167,32 @@ export class DpmsService {
           response
             .json()
             .then((responseResult: ITableResponse) => {
-              const filteredEntries = responseResult.results.filter(
-                (entry: { displayName: string }) => entry.displayName
-              );
+              const filteredEntries =
+                responseResult &&
+                responseResult.results &&
+                responseResult.results.filter(
+                  (entry: { displayName: string }) => entry.displayName
+                );
               const tableNames: string[] = [];
               const entryNames: string[] = [];
               const updatedTableDetails: { [key: string]: string } = {};
-              filteredEntries.forEach(
-                (entry: {
-                  displayName: string;
-                  relativeResourceName: string;
-                  description: string;
-                }) => {
-                  tableNames.push(entry.displayName);
-                  entryNames.push(entry.relativeResourceName);
-                  const description = entry.description || 'None';
-                  updatedTableDetails[entry.displayName] = description;
-                }
-              );
+              if (filteredEntries !== undefined) {
+                filteredEntries &&
+                  filteredEntries.forEach(
+                    (entry: {
+                      displayName: string;
+                      relativeResourceName: string;
+                      description: string;
+                    }) => {
+                      tableNames.push(entry.displayName);
+                      entryNames.push(entry.relativeResourceName);
+                      const description = entry.description || 'None';
+                      updatedTableDetails[entry.displayName] = description;
+                    }
+                  );
+              } else {
+                setTotalDatabases(totalDatabases - 1 || 0);
+              }
               setEntries(entryNames);
               setTableDescription(updatedTableDetails);
               setTotalTables(tableNames.length);
diff --git a/src/index.ts b/src/index.ts
index 8cf40388..f53516ea 100644
--- a/src/index.ts
+++ b/src/index.ts
@@ -563,7 +563,9 @@ const extension: JupyterFrontEndPlugin<void> = {
 
     const createAuthLoginComponentCommand = 'cloud-dataproc-settings:configure';
     commands.addCommand(createAuthLoginComponentCommand, {
-      label: 'Google BigQuery Settings',
+      label: bqFeature.enable_bigquery_integration
+        ? 'Google BigQuery Settings'
+        : 'Google Dataproc Settings',
       execute: () => {
         const content = new AuthLogin(
           app as JupyterLab,
diff --git a/src/jobs/jobServices.tsx b/src/jobs/jobServices.tsx
index 666a71f1..56a24f1f 100644
--- a/src/jobs/jobServices.tsx
+++ b/src/jobs/jobServices.tsx
@@ -266,6 +266,7 @@ export class JobService {
     setIsLoading: (value: boolean) => void,
     setjobsList: any,
     renderActions: (value: IRenderActionsData) => React.JSX.Element,
+    controllerSignal: any,
     nextPageToken?: string,
     previousJobsList?: object
   ) => {
@@ -275,7 +276,7 @@ export class JobService {
     const pageToken = nextPageToken ?? '';
     if (credentials) {
       loggedFetch(
-        `${DATAPROC}/projects/${credentials.project_id}/regions/${credentials.region_id}/jobs?pageSize=50&pageToken=${pageToken}&&clusterName=${clusterName}`,
+        `${DATAPROC}/projects/${credentials.project_id}/regions/${credentials.region_id}/jobs?pageSize=500&pageToken=${pageToken}&&clusterName=${clusterName}`,
         {
           headers: {
             'Content-Type': API_HEADER_CONTENT_TYPE,
@@ -298,10 +299,12 @@ export class JobService {
                 actions: React.JSX.Element;
               }[] = [];
               if (responseResult?.error?.code) {
-                toast.error(
-                  responseResult?.error?.message,
-                  toastifyCustomStyle
-                );
+                if (!toast.isActive('jobError')) {
+                  toast.error(responseResult?.error?.code, {
+                    ...toastifyCustomStyle,
+                    toastId: 'jobError'
+                  });
+                }
               }
               if (responseResult && responseResult.jobs) {
                 transformJobListData = responseResult.jobs.map((data: any) => {
@@ -346,12 +349,17 @@ export class JobService {
                 ...transformJobListData
               ];
 
-              if (responseResult.nextPageToken) {
+              if (
+                responseResult.nextPageToken &&
+                responseResult.jobs &&
+                !controllerSignal.aborted
+              ) {
                 this.listJobsAPIService(
                   clusterSelected,
                   setIsLoading,
                   setjobsList,
                   renderActions,
+                  controllerSignal,
                   responseResult.nextPageToken,
                   allJobsData
                 );
@@ -426,7 +434,12 @@ export class JobService {
         setClusterResponse(allClustersData);
       }
       if (formattedResponse?.error?.code) {
-        toast.error(formattedResponse?.error?.message, toastifyCustomStyle);
+        if (!toast.isActive('jobClusterError')) {
+          toast.error(formattedResponse?.error?.message, {
+            ...toastifyCustomStyle,
+            toastId: 'jobClusterError'
+          });
+        }
       }
     } catch (error) {
       DataprocLoggingService.log('Error listing clusters', LOG_LEVEL.ERROR);
diff --git a/src/jobs/jobs.tsx b/src/jobs/jobs.tsx
index 1b9015d2..a8ceb141 100644
--- a/src/jobs/jobs.tsx
+++ b/src/jobs/jobs.tsx
@@ -97,6 +97,7 @@ function JobComponent({
   setSelectedJobClone,
   fromPage
 }: any) {
+  const controller = useRef(new AbortController());
   const [jobsList, setjobsList] = useState([]);
   const [jobSelected, setjobSelected] = useState<string>('');
   const [isLoading, setIsLoading] = useState(true);
@@ -194,11 +195,14 @@ function JobComponent({
   };
 
   const listJobsAPI = async () => {
+    controller.current.abort(); 
+    controller.current = new AbortController(); 
     await JobService.listJobsAPIService(
       clusterSelected,
       setIsLoading,
       setjobsList,
-      renderActions
+      renderActions,
+      controller.current.signal 
     );
   };
 
@@ -286,9 +290,13 @@ function JobComponent({
       listClustersAPI();
     }
     return () => {
-      pollingJobs(listJobsAPI, true);
+      controller.current.abort();
+      if (timer.current) {
+        clearTimeout(timer.current);
+      }
     };
   }, [pollingDisable, detailedJobView]);
+  
   useEffect(() => {
     if (!detailedJobView && !isLoading) {
       pollingJobs(listJobsAPI, pollingDisable);
diff --git a/src/jobs/submitJob.tsx b/src/jobs/submitJob.tsx
index afffaf79..8a6c3817 100644
--- a/src/jobs/submitJob.tsx
+++ b/src/jobs/submitJob.tsx
@@ -34,7 +34,6 @@ import {
 } from '../utils/const';
 import errorIcon from '../../style/icons/error_icon.svg';
 import { Input } from '../controls/MuiWrappedInput';
-import { Select } from '../controls/MuiWrappedSelect';
 import { DropdownProps } from 'semantic-ui-react';
 import { Autocomplete, TextField } from '@mui/material';
 import { MuiChipsInput } from 'mui-chips-input';
@@ -60,16 +59,16 @@ function jobKey(selectedJobClone: any) {
   return jobKeys;
 }
 function jobTypeFunction(jobKey: string) {
-  let jobType = 'spark';
+  let jobType = 'Spark';
   switch (jobKey) {
     case 'sparkRJob':
-      jobType = 'sparkR';
+      jobType = 'SparkR';
       return jobType;
     case 'pysparkJob':
-      jobType = 'pySpark';
+      jobType = 'PySpark';
       return jobType;
     case 'sparkSqlJob':
-      jobType = 'sparkSql';
+      jobType = 'SparkSql';
       return jobType;
     default:
       return jobType;
@@ -138,7 +137,7 @@ function SubmitJob({
   let pythonFileUris: string[] = [];
   let maxFailuresPerHour = '';
   let mainRFileUri = '';
-  let jobType = 'spark';
+  let jobType = 'Spark';
   let mainPythonFileUri = '';
   let queryFileUri = '';
   let queryType = '';
@@ -229,7 +228,6 @@ function SubmitJob({
     useState(false);
   const [argumentsDuplicateValidation, setArgumentsDuplicateValidation] =
     useState(false);
-  const [labelFocused, setLabelFocused] = useState(false);
   const handleCancelJobButton = () => {
     setSubmitJobView(false);
   };
@@ -245,14 +243,15 @@ function SubmitJob({
   };
 
   const handleClusterSelected = (data: DropdownProps | null) => {
-    setClusterSelected(data!.toString());
+    if (data !== null) {
+      setClusterSelected(data!.toString());
+    }
   };
 
-  const handleJobTypeSelected = (
-    event: React.SyntheticEvent<HTMLElement, Event>,
-    data: DropdownProps
-  ) => {
-    setJobTypeSelected(data.value!.toString());
+  const handleJobTypeSelected = (data: DropdownProps | null) => {
+    if (data !== null) {
+      setJobTypeSelected(data!.toString());
+    }
     setFileSelected([]);
     setJarFileSelected([]);
     setAdditionalPythonFileSelected([]);
@@ -269,7 +268,9 @@ function SubmitJob({
     event: SyntheticEvent<Element, Event>,
     data: DropdownProps | null
   ) => {
-    setQuerySourceSelected(data!.toString());
+    if (data !== null) {
+      setQuerySourceSelected(data!.toString());
+    }
   };
   interface IClusterData {
     clusterName: string;
@@ -287,14 +288,9 @@ function SubmitJob({
     const keyLabelStructure = transformClusterListData.map(
       (obj: { clusterName: string }) => obj.clusterName
     );
-
     setClusterList(keyLabelStructure);
-    const jobTypeData = [
-      { key: 'spark', value: 'spark', text: 'Spark' },
-      { key: 'sparkR', value: 'sparkR', text: 'SparkR' },
-      { key: 'sparkSql', value: 'sparkSql', text: 'SparkSql' },
-      { key: 'pySpark', value: 'pySpark', text: 'PySpark' }
-    ];
+    const jobTypeData = ['Spark', 'SparkR', 'SparkSql', 'PySpark'];
+
     setJobTypeList(jobTypeData);
     const querySourceData = ['Query file', 'Query text'];
     setQuerySourceTypeList(querySourceData);
@@ -329,10 +325,10 @@ function SubmitJob({
     archiveDuplicateValidation
   ]);
   const disableSubmitButtonIfInvalid = () => {
-    const isSparkJob = jobTypeSelected === 'spark';
-    const isSparkRJob = jobTypeSelected === 'sparkR';
-    const isPySparkJob = jobTypeSelected === 'pySpark';
-    const isSparkSqlJob = jobTypeSelected === 'sparkSql';
+    const isSparkJob = jobTypeSelected === 'Spark';
+    const isSparkRJob = jobTypeSelected === 'SparkR';
+    const isPySparkJob = jobTypeSelected === 'PySpark';
+    const isSparkSqlJob = jobTypeSelected === 'SparkSql';
     if (
       clusterSelected !== '' &&
       jobIdSelected !== '' &&
@@ -603,7 +599,7 @@ function SubmitJob({
           ...(labelObject && {
             labels: labelObject
           }),
-          ...(jobTypeSelected === 'pySpark' &&
+          ...(jobTypeSelected === 'PySpark' &&
             createPySparkPayload(
               mainPythonSelected,
               propertyObject,
@@ -613,7 +609,7 @@ function SubmitJob({
               argumentSelected,
               additionalPythonFileSelected
             )),
-          ...(jobTypeSelected === 'spark' &&
+          ...(jobTypeSelected === 'Spark' &&
             createSparkPayload(
               mainClassSelected,
               propertyObject,
@@ -622,14 +618,14 @@ function SubmitJob({
               jarFileSelected,
               argumentSelected
             )),
-          ...(jobTypeSelected === 'sparkR' &&
+          ...(jobTypeSelected === 'SparkR' &&
             createSparkRPayload(
               mainRSelected,
               propertyObject,
               fileSelected,
               argumentSelected
             )),
-          ...(jobTypeSelected === 'sparkSql' &&
+          ...(jobTypeSelected === 'SparkSql' &&
             createSparkSqlPayload(
               propertyObject,
               jarFileSelected,
@@ -721,12 +717,6 @@ function SubmitJob({
     setArgumentSelected(listOfFiles);
     handleDuplicateValidation(setDuplicateValidation, listOfFiles);
   };
-  const handleSelectFocus = () => {
-    setLabelFocused(true);
-  };
-  const handleBlur = () => {
-    setLabelFocused(false);
-  };
   return (
     <div>
       <div className="cluster-details-header">
@@ -785,28 +775,16 @@ function SubmitJob({
         )}
 
         <div className="select-text-overlay">
-          <label
-            className={
-              labelFocused
-                ? 'select-title-text label-focused'
-                : 'select-title-text'
-            }
-            htmlFor="metastore-project"
-          >
-            Job type*
-          </label>
-          <Select
+          <Autocomplete
             className="project-region-select"
-            search
-            onChange={handleJobTypeSelected}
+            onChange={(_event, val) => handleJobTypeSelected(val)}
             options={jobTypeList}
             value={jobTypeSelected}
-            onFocus={handleSelectFocus}
-            onBlur={handleBlur}
+            renderInput={params => <TextField {...params} label=" Job type*" />}
           />
         </div>
 
-        {jobTypeSelected === 'sparkSql' && (
+        {jobTypeSelected === 'SparkSql' && (
           <>
             <div className="select-text-overlay">
               <Autocomplete
@@ -816,8 +794,6 @@ function SubmitJob({
                 }
                 options={querySourceTypeList}
                 value={querySourceSelected}
-                onFocus={handleSelectFocus}
-                onBlur={handleBlur}
                 renderInput={params => (
                   <TextField {...params} label="Query source type*" />
                 )}
@@ -826,7 +802,7 @@ function SubmitJob({
           </>
         )}
         {querySourceSelected === 'Query file' &&
-          jobTypeSelected === 'sparkSql' && (
+          jobTypeSelected === 'SparkSql' && (
             <>
               <div className="select-text-overlay">
                 <Input
@@ -861,7 +837,7 @@ function SubmitJob({
             </>
           )}
         {querySourceSelected === 'Query text' &&
-          jobTypeSelected === 'sparkSql' && (
+          jobTypeSelected === 'SparkSql' && (
             <>
               <div className="select-text-overlay">
                 <Input
@@ -877,7 +853,7 @@ function SubmitJob({
               </div>
             </>
           )}
-        {jobTypeSelected === 'spark' && (
+        {jobTypeSelected === 'Spark' && (
           <>
             <div className="select-text-overlay">
               <Input
@@ -911,7 +887,7 @@ function SubmitJob({
             )}
           </>
         )}
-        {jobTypeSelected === 'sparkR' && (
+        {jobTypeSelected === 'SparkR' && (
           <>
             <div className="select-text-overlay">
               <Input
@@ -945,7 +921,7 @@ function SubmitJob({
             )}
           </>
         )}
-        {jobTypeSelected === 'pySpark' && (
+        {jobTypeSelected === 'PySpark' && (
           <>
             <div className="select-text-overlay">
               <Input
@@ -979,7 +955,7 @@ function SubmitJob({
             )}
           </>
         )}
-        {jobTypeSelected === 'pySpark' && (
+        {jobTypeSelected === 'PySpark' && (
           <>
             <div className="select-text-overlay">
               <MuiChipsInput
@@ -1017,7 +993,7 @@ function SubmitJob({
             )}
           </>
         )}
-        {jobTypeSelected !== 'sparkR' && (
+        {jobTypeSelected !== 'SparkR' && (
           <>
             <div className="select-text-overlay">
               <MuiChipsInput
@@ -1058,7 +1034,7 @@ function SubmitJob({
             )}
           </>
         )}
-        {jobTypeSelected !== 'sparkSql' && (
+        {jobTypeSelected !== 'SparkSql' && (
           <>
             <div className="select-text-overlay">
               <MuiChipsInput
@@ -1099,7 +1075,7 @@ function SubmitJob({
             )}
           </>
         )}
-        {(jobTypeSelected === 'spark' || jobTypeSelected === 'pySpark') && (
+        {(jobTypeSelected === 'Spark' || jobTypeSelected === 'PySpark') && (
           <>
             <div className="select-text-overlay">
               <MuiChipsInput
@@ -1140,7 +1116,7 @@ function SubmitJob({
             )}
           </>
         )}
-        {jobTypeSelected !== 'sparkSql' && (
+        {jobTypeSelected !== 'SparkSql' && (
           <>
             <div className="select-text-overlay">
               <MuiChipsInput
@@ -1168,7 +1144,7 @@ function SubmitJob({
           </>
         )}
         {querySourceSelected === 'Query file' &&
-          jobTypeSelected === 'sparkSql' && (
+          jobTypeSelected === 'SparkSql' && (
             <>
               <div className="submit-job-label-header">Query parameters</div>
               <LabelProperties
diff --git a/src/login/configSelection.tsx b/src/login/configSelection.tsx
index ace38854..7c62aaa0 100644
--- a/src/login/configSelection.tsx
+++ b/src/login/configSelection.tsx
@@ -17,7 +17,7 @@
 
 import React, { useState, useEffect } from 'react';
 import { LabIcon } from '@jupyterlab/ui-components';
-import settingsIcon from '../../style/icons/settings_icon.svg';
+import googleCloudIcon from '../../style/icons/google-cloud.svg';
 import {
   API_HEADER_BEARER,
   API_HEADER_CONTENT_TYPE,
@@ -75,9 +75,9 @@ function ConfigSelection({
   launcher,
   settingRegistry
 }: IConfigSelectionProps) {
-  const Iconsettings = new LabIcon({
-    name: 'launcher:settings_icon',
-    svgstr: settingsIcon
+  const IconGoogleCloud = new LabIcon({
+    name: 'launcher:google_cloud_icon',
+    svgstr: googleCloudIcon
   });
 
   const [bigQueryFeatureEnable, setbigQueryFeatureEnable] = useState(false);
@@ -257,7 +257,10 @@ function ConfigSelection({
         <div className="settings-component">
           <div className="settings-overlay">
             <div>
-              <Iconsettings.react tag="div" className="logo-alignment-style" />
+              <IconGoogleCloud.react
+                tag="div"
+                className="logo-alignment-style"
+              />
             </div>
             <div className="settings-text">Settings</div>
           </div>
@@ -375,9 +378,10 @@ function ConfigSelection({
             </div>
           </div>
           <div>
+            <div className="dataproc-settings-header">Dataproc Settings </div>
             <div className="runtime-title-section">
               <div className="runtime-title-part">
-                Dataproc Serverless Runtime Templates
+                Serverless Runtime Templates
               </div>
               <div
                 className="expand-icon"
diff --git a/src/runtime/createRunTime.tsx b/src/runtime/createRunTime.tsx
index 9d54f5e7..a54c5c37 100644
--- a/src/runtime/createRunTime.tsx
+++ b/src/runtime/createRunTime.tsx
@@ -188,8 +188,11 @@ function CreateRunTime({
     region,
     servicesSelected
   ]);
+
   useEffect(() => {
-    listSubNetworksAPI(networkSelected);
+    if (networkSelected !== '') {
+      listSubNetworksAPI(networkSelected);
+    }
   }, [networkSelected]);
 
   const displayUserInfo = async () => {
@@ -387,7 +390,9 @@ function CreateRunTime({
   const listNetworksAPI = async () => {
     await RunTimeSerive.listNetworksAPIService(
       setNetworklist,
-      setNetworkSelected
+      setNetworkSelected,
+      selectedRuntimeClone,
+      setIsloadingNetwork
     );
   };
 
@@ -395,7 +400,9 @@ function CreateRunTime({
     await RunTimeSerive.listSubNetworksAPIService(
       subnetwork,
       setSubNetworklist,
-      setSubNetworkSelected
+      setSubNetworkSelected,
+      selectedRuntimeClone,
+      setIsloadingNetwork
     );
   };
 
@@ -453,7 +460,9 @@ function CreateRunTime({
   };
 
   const handleServiceSelected = (data: string | null) => {
-    setServicesSelected(data!.toString());
+    if (data !== null) {
+      setServicesSelected(data!.toString());
+    }
   };
   const handleIdleSelected = (event: React.ChangeEvent<HTMLInputElement>) => {
     const inputValue = event.target.value;
@@ -498,10 +507,12 @@ function CreateRunTime({
   };
 
   const handleNetworkChange = async (data: DropdownProps | null) => {
-    setNetworkSelected(data!.toString());
-    setSubNetworkSelected(defaultValue);
-    await listSubNetworksAPI(data!.toString());
-    await handleProjectIdChange(projectId, data!.toString());
+    if (data !== null) {
+      setNetworkSelected(data!.toString());
+      setSubNetworkSelected(defaultValue);
+      await listSubNetworksAPI(data!.toString());
+      await handleProjectIdChange(projectId, data!.toString());
+    }
   };
 
   const handleNetworkSharedVpcRadioChange = () => {
@@ -514,11 +525,15 @@ function CreateRunTime({
     setSharedvpcSelected('');
   };
   const handleSubNetworkChange = (data: string | null) => {
-    setSubNetworkSelected(data!.toString());
+    if (data !== null) {
+      setSubNetworkSelected(data!.toString());
+    }
   };
   const handleSharedSubNetwork = async (data: string | null) => {
-    setSharedvpcSelected(data!.toString());
-    await handleProjectIdChange(projectId, data!.toString());
+    if (data !== null) {
+      setSharedvpcSelected(data!.toString());
+      await handleProjectIdChange(projectId, data!.toString());
+    }
   };
   const handleCancelButton = async () => {
     setOpenCreateTemplate(false);
@@ -850,7 +865,7 @@ function CreateRunTime({
       {configLoading && !loggedIn && !configError && !loginError && (
         <div className="spin-loader-main">
           <CircularProgress
-            className = "spin-loader-custom-style"
+            className="spin-loader-custom-style"
             size={18}
             aria-label="Loading Spinner"
             data-testid="loader"
@@ -1111,9 +1126,10 @@ function CreateRunTime({
                       </div>
                     </div>
                   )}
-                {selectedNetworkRadio === 'projectNetwork' &&
-                  networkList.length !== 0 &&
-                  subNetworkList.length === 0 && (
+                {!isloadingNetwork &&
+                  selectedNetworkRadio === 'projectNetwork' &&
+                  networkSelected !== '' &&
+                  subNetworkSelected === '' && (
                     <div className="error-key-parent">
                       <iconError.react
                         tag="div"
diff --git a/src/runtime/runtimeService.tsx b/src/runtime/runtimeService.tsx
index 1a0be39e..614586af 100644
--- a/src/runtime/runtimeService.tsx
+++ b/src/runtime/runtimeService.tsx
@@ -396,7 +396,6 @@ export class RunTimeSerive {
               setNetworkSelected(transformedNetworkSelected);
               setSubNetworkSelected(subnetwork);
               setDefaultValue(subnetwork);
-              setIsloadingNetwork(false);
               if (responseResult?.error?.code) {
                 toast.error(
                   responseResult?.error?.message,
@@ -410,7 +409,6 @@ export class RunTimeSerive {
             });
         })
         .catch((err: Error) => {
-          setIsloadingNetwork(false);
           DataprocLoggingService.log(
             'Error selecting Network',
             LOG_LEVEL.ERROR
@@ -454,8 +452,11 @@ export class RunTimeSerive {
   };
   static listNetworksAPIService = async (
     setNetworklist: (value: string[]) => void,
-    setNetworkSelected: (value: string) => void
+    setNetworkSelected: (value: string) => void,
+    selectedRuntimeClone: any,
+    setIsloadingNetwork: (value: boolean) => void
   ) => {
+    setIsloadingNetwork(true);
     const { COMPUTE } = await gcpServiceUrls;
     try {
       const response = await authenticatedFetch({
@@ -476,7 +477,9 @@ export class RunTimeSerive {
       });
 
       setNetworklist(transformedNetworkList);
-      setNetworkSelected(transformedNetworkList[0]);
+      if (selectedRuntimeClone === undefined) {
+        setNetworkSelected(transformedNetworkList[0]);
+      }
     } catch (error) {
       DataprocLoggingService.log('Error listing Networks', LOG_LEVEL.ERROR);
       toast.error(`Error listing Networks : ${error}`, toastifyCustomStyle);
@@ -635,8 +638,11 @@ export class RunTimeSerive {
   static listSubNetworksAPIService = async (
     subnetwork: string,
     setSubNetworklist: (value: string[]) => void,
-    setSubNetworkSelected: (value: string) => void
+    setSubNetworkSelected: (value: string) => void,
+    selectedRuntimeClone: any,
+    setIsloadingNetwork: (value: boolean) => void
   ) => {
+    setIsloadingNetwork(true);
     const credentials = await authApi();
     const { COMPUTE } = await gcpServiceUrls;
     if (credentials) {
@@ -673,7 +679,10 @@ export class RunTimeSerive {
                   (data: { name: string }) => data.name
                 );
                 setSubNetworklist(transformedServiceList);
-                setSubNetworkSelected(transformedServiceList[0]);
+                if (selectedRuntimeClone === undefined) {
+                  setSubNetworkSelected(transformedServiceList[0]);
+                }
+                setIsloadingNetwork(false);
                 if (responseResult?.error?.code) {
                   toast.error(
                     responseResult?.error?.message,
@@ -691,6 +700,7 @@ export class RunTimeSerive {
             'Error listing subNetworks',
             LOG_LEVEL.ERROR
           );
+          setIsloadingNetwork(false);
           toast.error(
             `Error listing subNetworks : ${err}`,
             toastifyCustomStyle
diff --git a/src/scheduler/schedulerServices.tsx b/src/scheduler/schedulerServices.tsx
index c940b8b3..9d86e268 100644
--- a/src/scheduler/schedulerServices.tsx
+++ b/src/scheduler/schedulerServices.tsx
@@ -132,7 +132,12 @@ export class SchedulerService {
         setClusterList(keyLabelStructure);
       }
       if (formattedResponse?.error?.code) {
-        toast.error(formattedResponse?.error?.message, toastifyCustomStyle);
+        if (!toast.isActive('clusterError')) {
+          toast.error(formattedResponse?.error?.message, {
+            ...toastifyCustomStyle,
+            toastId: 'clusterError'
+          });
+        }
       }
     } catch (error) {
       DataprocLoggingService.log('Error listing clusters', LOG_LEVEL.ERROR);
@@ -190,7 +195,12 @@ export class SchedulerService {
         setServerlessList(keyLabelStructure);
       }
       if (formattedResponse?.error?.code) {
-        toast.error(formattedResponse?.error?.message, toastifyCustomStyle);
+        if (!toast.isActive('sessionTemplateError')) {
+          toast.error(formattedResponse?.error?.message, {
+            ...toastifyCustomStyle,
+            toastId: 'sessionTemplateError'
+          });
+        }
       }
     } catch (error) {
       DataprocLoggingService.log(
@@ -455,7 +465,7 @@ export class SchedulerService {
     setRedListDates([]);
     setGreenListDates([]);
     setDarkGreenListDates([]);
-    try {     
+    try {
       const data: any = await requestAPI(
         `dagRun?composer=${composerName}&dag_id=${dagId}&start_date=${start_date}&end_date=${end_date}&offset=${offset}`
       );
@@ -666,24 +676,6 @@ export class SchedulerService {
       }
     }
   };
-  static handleDownloadSchedulerAPIService = async (
-    composerSelected: string,
-    jobid: string,
-    bucketName: string
-  ) => {
-    try {
-      const serviceURL = `dagDownload?composer=${composerSelected}&dag_id=${jobid}&bucket_name=${bucketName}`;
-      const formattedResponse: any = await requestAPI(serviceURL);
-      if (formattedResponse.status === 0) {
-        toast.success(`${jobid} downloaded successfully`, toastifyCustomStyle);
-      } else {
-        toast.error(`Failed to download the ${jobid}`, toastifyCustomStyle);
-      }
-    } catch (error) {
-      DataprocLoggingService.log('Error in Download api', LOG_LEVEL.ERROR);
-      toast.error(`Error in Download api : ${error}`, toastifyCustomStyle);
-    }
-  };
   static handleDownloadOutputNotebookAPIService = async (
     dagRunId: string,
     bucketName: string,
diff --git a/src/sessions/sessionService.tsx b/src/sessions/sessionService.tsx
index 56debac4..7e48e030 100644
--- a/src/sessions/sessionService.tsx
+++ b/src/sessions/sessionService.tsx
@@ -241,7 +241,12 @@ export class SessionService {
         setIsLoading(false);
       }
       if (formattedResponse?.error?.code) {
-        toast.error(formattedResponse?.error?.message, toastifyCustomStyle);
+        if (!toast.isActive('sessionError')) {
+          toast.error(formattedResponse?.error?.message, {
+            ...toastifyCustomStyle,
+            toastId: 'sessionError'
+          });
+        }
         setIsLoading(false);
       }
     } catch (error) {
diff --git a/src/utils/const.ts b/src/utils/const.ts
index 0acaa3de..285f03c6 100644
--- a/src/utils/const.ts
+++ b/src/utils/const.ts
@@ -16,7 +16,7 @@
  */
 
 import { requestAPI } from '../handler/handler';
-export const VERSION_DETAIL = '0.1.77';
+export const VERSION_DETAIL = '0.1.78';
 export const CREATE_CLUSTER_URL =
   'https://console.cloud.google.com/dataproc/clusters';
 export const CREATE_BATCH_URL =
diff --git a/src/utils/tableData.tsx b/src/utils/tableData.tsx
index d27ec130..f93aa3f2 100644
--- a/src/utils/tableData.tsx
+++ b/src/utils/tableData.tsx
@@ -31,26 +31,28 @@ function TableData({
   tableDataCondition,
   fromPage
 }: any) {
-  const [listDagRunHeight, setListDagRunHeight] = useState(window.innerHeight - 505);
+  const [listDagRunHeight, setListDagRunHeight] = useState(
+    window.innerHeight - 505
+  );
 
   function handleUpdateHeight() {
     let updateHeight = window.innerHeight - 505;
     setListDagRunHeight(updateHeight);
   }
 
-   // Debounce the handleUpdateHeight function
-   const debouncedHandleUpdateHeight = handleDebounce(handleUpdateHeight, 500);
+  // Debounce the handleUpdateHeight function
+  const debouncedHandleUpdateHeight = handleDebounce(handleUpdateHeight, 500);
+
+  // Add event listener for window resize using useEffect
+  useEffect(() => {
+    window.addEventListener('resize', debouncedHandleUpdateHeight);
+
+    // Cleanup function to remove event listener on component unmount
+    return () => {
+      window.removeEventListener('resize', debouncedHandleUpdateHeight);
+    };
+  }, []);
 
-   // Add event listener for window resize using useEffect
-   useEffect(() => {
-     window.addEventListener('resize', debouncedHandleUpdateHeight);
- 
-     // Cleanup function to remove event listener on component unmount
-     return () => {
-       window.removeEventListener('resize', debouncedHandleUpdateHeight);
-     };
-   }, []);
-  
   const displayData = page ? page : rows;
 
   return (
@@ -78,7 +80,13 @@ function TableData({
         style={fromPage === 'Dag Runs' ? { maxHeight: listDagRunHeight } : null}
       >
         {isLoading ? (
-          <div className="spin-loader">
+          <div
+            className={
+              fromPage === 'Preview'
+                ? 'spin-loader-preview-data'
+                : 'spin-loader'
+            }
+          >
             <CircularProgress
               className="spin-loader-custom-style"
               size={18}
diff --git a/style/authLogin.css b/style/authLogin.css
index a66a1914..f26990ac 100644
--- a/style/authLogin.css
+++ b/style/authLogin.css
@@ -253,11 +253,19 @@
   color: var(--jp-brand-color0) !important;
 }
 
+.dataproc-settings-header {
+  padding-bottom: 15px;
+  font-style: normal;
+  font-weight: 500;
+  font-size: 20px;
+  color: var(--jp-ui-font-color0);
+}
+
 .runtime-title-section {
   display: flex;
   justify-content: space-between;
   color: var(--jp-ui-font-color0);
-  font-size: 20px;
+  font-size: 16px;
   font-style: normal;
   font-weight: 500;
   margin-right: 20px;
@@ -290,7 +298,8 @@
   margin-right: 25px;
 }
 
-.project-header, .bigquery-region-header {
+.project-header,
+.bigquery-region-header {
   padding-top: 40px;
   padding-left: 5px;
   font-style: normal;
diff --git a/style/cluster.css b/style/cluster.css
index 5ff7a76b..320891d2 100644
--- a/style/cluster.css
+++ b/style/cluster.css
@@ -111,6 +111,13 @@
   margin-left: 10px;
 }
 
+.spin-loader-preview-data {
+  display: flex;
+  margin-top: 10px;
+  margin-left: 10px;
+  width: 150px;
+}
+
 .spin-loader-main {
   display: flex;
   max-height: 100%;
diff --git a/style/submitJob.css b/style/submitJob.css
index 65d47732..ab8b8b62 100644
--- a/style/submitJob.css
+++ b/style/submitJob.css
@@ -89,7 +89,7 @@
   color: #000000;
   position: absolute;
   background-color: #fff;
-  top: 2px;
+  top: -5px;
   padding: 0px 3px;
   left: 7px;
   z-index: 3;


Test patch:
diff --git a/dataproc_jupyter_plugin/tests/test_airflow.py b/dataproc_jupyter_plugin/tests/test_airflow.py
new file mode 100644
index 00000000..19f0c533
--- /dev/null
+++ b/dataproc_jupyter_plugin/tests/test_airflow.py
@@ -0,0 +1,321 @@
+# Copyright 2024 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import json
+import subprocess
+from unittest.mock import Mock
+
+import pytest
+import requests
+from google.cloud import jupyter_config
+
+from dataproc_jupyter_plugin import credentials
+from dataproc_jupyter_plugin.services import airflow
+
+
+async def mock_credentials():
+    return {
+        "project_id": "credentials-project",
+        "project_number": 12345,
+        "region_id": "mock-region",
+        "access_token": "mock-token",
+        "config_error": 0,
+        "login_error": 0,
+    }
+
+
+def mock_get(api_endpoint, headers=None):
+    response = Mock()
+    response.status_code = 200
+    response.json.return_value = {
+        "api_endpoint": api_endpoint,
+        "headers": headers,
+    }
+    return response
+
+
+async def mock_config(field_name):
+    return None
+
+
+async def mock_get_airflow_uri(self, composer_name):
+    return "https://mock_airflow_uri", "mock_bucket"
+
+
+async def test_list_jobs(monkeypatch, jp_fetch):
+    monkeypatch.setattr(credentials, "get_cached", mock_credentials)
+    monkeypatch.setattr(requests, "get", mock_get)
+    monkeypatch.setattr(jupyter_config, "async_get_gcloud_config", mock_config)
+    monkeypatch.setattr(airflow.Client, "get_airflow_uri", mock_get_airflow_uri)
+    mock_composer = "mock_url"
+    response = await jp_fetch(
+        "dataproc-plugin",
+        "dagList",
+        params={"composer": mock_composer},
+    )
+
+    assert response.code == 200
+    payload = json.loads(response.body)
+    assert (
+        payload[0]["api_endpoint"]
+        == "https://mock_airflow_uri/api/v1/dags?tags=dataproc_jupyter_plugin"
+    )
+    assert payload[0]["headers"]["Authorization"] == f"Bearer mock-token"
+
+
+async def test_list_dag_with_invalid_credentials(monkeypatch, jp_fetch):
+    async def mock_credentials():
+        return {}
+
+    monkeypatch.setattr(credentials, "get_cached", mock_credentials)
+    monkeypatch.setattr(requests, "get", mock_get)
+    monkeypatch.setattr(jupyter_config, "async_get_gcloud_config", mock_config)
+    monkeypatch.setattr(airflow.Client, "get_airflow_uri", mock_get_airflow_uri)
+    response = await jp_fetch(
+        "dataproc-plugin",
+        "dagList",
+        params={"project_id": "project_id"},
+    )
+    assert response.code == 200
+    payload = json.loads(response.body)
+    assert payload == {"error": "Missing required credentials"}
+
+
+@pytest.mark.parametrize("returncode, expected_result", [(0, 0)])
+async def test_delete_job(monkeypatch, returncode, expected_result, jp_fetch):
+    def mock_popen(returncode=0):
+        def _mock_popen(*args, **kwargs):
+            mock_process = Mock()
+            mock_process.communicate.return_value = (b"output", b"")
+            mock_process.returncode = returncode
+            return mock_process
+
+        return _mock_popen
+
+    monkeypatch.setattr(credentials, "get_cached", mock_credentials)
+    monkeypatch.setattr(requests, "get", mock_get)
+    monkeypatch.setattr(jupyter_config, "async_get_gcloud_config", mock_config)
+    monkeypatch.setattr(airflow.Client, "get_airflow_uri", mock_get_airflow_uri)
+    monkeypatch.setattr(subprocess, "Popen", mock_popen(returncode))
+
+    mock_composer = "mock_composer"
+    mock_dag_id = "mock_dag_id"
+    mock_from_page = "mock_from_page"
+    response = await jp_fetch(
+        "dataproc-plugin",
+        "dagDelete",
+        params={
+            "composer": mock_composer,
+            "dag_id": mock_dag_id,
+            "from_page": mock_from_page,
+        },
+    )
+    assert response.code == 200
+    payload = json.loads(response.body)
+    assert payload["status"] == 0
+
+
+async def test_update_job(monkeypatch, jp_fetch):
+
+    def mock_patch(url, json, headers):
+        mock_resp = Mock()
+        mock_resp.status_code = 200 if json["is_paused"] is False else 400
+        return mock_resp
+
+    async def mock_config(config_field):
+        return ""
+
+    monkeypatch.setattr(credentials, "get_cached", mock_credentials)
+    monkeypatch.setattr(requests, "patch", mock_patch)
+    monkeypatch.setattr(jupyter_config, "async_get_gcloud_config", mock_config)
+    monkeypatch.setattr(airflow.Client, "get_airflow_uri", mock_get_airflow_uri)
+    mock_composer = "composer"
+    mock_dag_id = "mock_dag_id"
+    mock_status = "true"
+    response = await jp_fetch(
+        "dataproc-plugin",
+        "dagUpdate",
+        params={
+            "composer": mock_composer,
+            "dag_id": mock_dag_id,
+            "status": mock_status,
+        },
+    )
+    assert response.code == 200
+    payload = json.loads(response.body)
+    assert payload["status"] == 0
+
+
+async def test_list_dag_run(monkeypatch, jp_fetch):
+    monkeypatch.setattr(credentials, "get_cached", mock_credentials)
+    monkeypatch.setattr(requests, "get", mock_get)
+    monkeypatch.setattr(airflow.Client, "get_airflow_uri", mock_get_airflow_uri)
+    monkeypatch.setattr(jupyter_config, "async_get_gcloud_config", mock_config)
+
+    mock_composer = "mock_url"
+    mock_dag_id = "mock_dag_id"
+    mock_start_date = "mock_start_date"
+    mock_offset = "mock_offset"
+    mock_end_date = "mock_end_date"
+    response = await jp_fetch(
+        "dataproc-plugin",
+        "dagRun",
+        params={
+            "composer": mock_composer,
+            "dag_id": mock_dag_id,
+            "start_date": mock_start_date,
+            "offset": mock_offset,
+            "end_date": mock_end_date,
+        },
+    )
+    assert response.code == 200
+    payload = json.loads(response.body)
+
+    assert (
+        payload["api_endpoint"]
+        == f"https://mock_airflow_uri/api/v1/dags/mock_dag_id/dagRuns?execution_date_gte={mock_start_date}&execution_date_lte={mock_end_date}&offset={mock_offset}"
+    )
+    assert payload["headers"]["Authorization"] == f"Bearer mock-token"
+
+
+async def test_list_dag_run_task_logs(monkeypatch, jp_fetch):
+    def mock_get(url, headers):
+        mock_resp = Mock()
+        mock_resp.status_code = 200
+        mock_resp.text = "mock log content"
+        return mock_resp
+
+    monkeypatch.setattr(credentials, "get_cached", mock_credentials)
+    monkeypatch.setattr(requests, "get", mock_get)
+    monkeypatch.setattr(airflow.Client, "get_airflow_uri", mock_get_airflow_uri)
+    monkeypatch.setattr(jupyter_config, "async_get_gcloud_config", mock_config)
+
+    mock_composer = "mock_url"
+    mock_dag_id = "mock_dag_id"
+    mock_dag_run_id = "mock_dag_run_id"
+    mock_task_id = "mock_task_id"
+    mock_task_try = "mock_task_try"
+    response = await jp_fetch(
+        "dataproc-plugin",
+        "dagRunTaskLogs",
+        params={
+            "composer": mock_composer,
+            "dag_id": mock_dag_id,
+            "dag_run_id": mock_dag_run_id,
+            "task_id": mock_task_id,
+            "task_try_number": mock_task_try,
+        },
+    )
+    assert response.code == 200
+    payload = json.loads(response.body)
+    assert payload["content"] == "mock log content"
+
+
+async def test_list_dag_run_task(monkeypatch, jp_fetch):
+
+    monkeypatch.setattr(credentials, "get_cached", mock_credentials)
+    monkeypatch.setattr(requests, "get", mock_get)
+    monkeypatch.setattr(airflow.Client, "get_airflow_uri", mock_get_airflow_uri)
+    monkeypatch.setattr(jupyter_config, "async_get_gcloud_config", mock_config)
+
+    mock_composer = "mock_url"
+    mock_dag_id = "mock_dag_id"
+    mock_dag_run_id = "mock_dag_run_id"
+    response = await jp_fetch(
+        "dataproc-plugin",
+        "dagRunTask",
+        params={
+            "composer": mock_composer,
+            "dag_id": mock_dag_id,
+            "dag_run_id": mock_dag_run_id,
+        },
+    )
+    assert response.code == 200
+    payload = json.loads(response.body)
+    assert (
+        payload["api_endpoint"]
+        == f"https://mock_airflow_uri/api/v1/dags/{mock_dag_id}/dagRuns/{mock_dag_run_id}/taskInstances"
+    )
+    assert payload["headers"]["Authorization"] == f"Bearer mock-token"
+
+
+async def test_edit_jobs(monkeypatch, jp_fetch):
+    monkeypatch.setattr(credentials, "get_cached", mock_credentials)
+    monkeypatch.setattr(requests, "get", mock_get)
+
+    mock_bucket_name = "mock_url"
+    mock_dag_id = "mock_dag_id"
+    response = await jp_fetch(
+        "dataproc-plugin",
+        "editJobScheduler",
+        params={"bucket_name": mock_bucket_name, "dag_id": mock_dag_id},
+    )
+    assert response.code == 200
+
+
+async def test_list_import_errors(monkeypatch, jp_fetch):
+    monkeypatch.setattr(credentials, "get_cached", mock_credentials)
+    monkeypatch.setattr(requests, "get", mock_get)
+    monkeypatch.setattr(airflow.Client, "get_airflow_uri", mock_get_airflow_uri)
+    monkeypatch.setattr(jupyter_config, "async_get_gcloud_config", mock_config)
+
+    mock_composer = "mock_composer"
+    response = await jp_fetch(
+        "dataproc-plugin",
+        "importErrorsList",
+        params={"composer": mock_composer},
+    )
+    assert response.code == 200
+    payload = json.loads(response.body)
+    assert (
+        payload["api_endpoint"]
+        == f"https://mock_airflow_uri/api/v1/importErrors?order_by=-import_error_id"
+    )
+    assert payload["headers"]["Authorization"] == f"Bearer mock-token"
+
+
+async def test_dag_trigger(monkeypatch, jp_fetch):
+    def mock_post(api_endpoint, headers=None, json=None):
+        response = Mock()
+        response.status_code = 200
+        response.json.return_value = {
+            "results": [
+                {
+                    "api_endpoint": api_endpoint,
+                    "headers": headers,
+                    "json": json,
+                }
+            ]
+        }
+        return response
+
+    monkeypatch.setattr(credentials, "get_cached", mock_credentials)
+    monkeypatch.setattr(requests, "post", mock_post)
+    monkeypatch.setattr(airflow.Client, "get_airflow_uri", mock_get_airflow_uri)
+    monkeypatch.setattr(jupyter_config, "async_get_gcloud_config", mock_config)
+
+    mock_composer = "mock_url"
+    mock_dag_id = "mock_dag_id"
+    response = await jp_fetch(
+        "dataproc-plugin",
+        "triggerDag",
+        params={"dag_id": mock_dag_id, "composer": mock_composer},
+    )
+    assert response.code == 200
+    payload = json.loads(response.body)["results"][0]
+    assert (
+        payload["api_endpoint"]
+        == f"https://mock_airflow_uri/api/v1/dags/{mock_dag_id}/dagRuns"
+    )
+    assert payload["headers"]["Authorization"] == f"Bearer mock-token"
diff --git a/dataproc_jupyter_plugin/tests/test_bigquery.py b/dataproc_jupyter_plugin/tests/test_bigquery.py
index 1ac46112..64503b1e 100644
--- a/dataproc_jupyter_plugin/tests/test_bigquery.py
+++ b/dataproc_jupyter_plugin/tests/test_bigquery.py
@@ -13,10 +13,9 @@
 # limitations under the License.
 
 import json
-import requests
-
 from unittest.mock import Mock
 
+import requests
 from google.cloud import jupyter_config
 
 from dataproc_jupyter_plugin import credentials
@@ -43,8 +42,13 @@ def mock_get(api_endpoint, headers=None):
     return response
 
 
+async def mock_config(field_name):
+    return None
+
+
 async def test_list_datasets(monkeypatch, jp_fetch):
     monkeypatch.setattr(credentials, "get_cached", mock_credentials)
+    monkeypatch.setattr(jupyter_config, "async_get_gcloud_config", mock_config)
     monkeypatch.setattr(requests, "get", mock_get)
 
     mock_project_id = "mock-project-id"
@@ -68,6 +72,7 @@ async def mock_credentials():
         return {}
 
     monkeypatch.setattr(credentials, "get_cached", mock_credentials)
+    monkeypatch.setattr(jupyter_config, "async_get_gcloud_config", mock_config)
     monkeypatch.setattr(requests, "get", mock_get)
     response = await jp_fetch(
         "dataproc-plugin",
@@ -81,6 +86,7 @@ async def mock_credentials():
 
 async def test_list_tables(monkeypatch, jp_fetch):
     monkeypatch.setattr(credentials, "get_cached", mock_credentials)
+    monkeypatch.setattr(jupyter_config, "async_get_gcloud_config", mock_config)
     monkeypatch.setattr(requests, "get", mock_get)
 
     mock_dataset_id = "mock-dataset-id"
@@ -106,6 +112,7 @@ async def test_list_tables(monkeypatch, jp_fetch):
 
 async def test_dataset_info(monkeypatch, jp_fetch):
     monkeypatch.setattr(credentials, "get_cached", mock_credentials)
+    monkeypatch.setattr(jupyter_config, "async_get_gcloud_config", mock_config)
     monkeypatch.setattr(requests, "get", mock_get)
 
     mock_dataset_id = "mock-dataset-id"
@@ -129,6 +136,7 @@ async def test_dataset_info(monkeypatch, jp_fetch):
 
 async def test_table_info(monkeypatch, jp_fetch):
     monkeypatch.setattr(credentials, "get_cached", mock_credentials)
+    monkeypatch.setattr(jupyter_config, "async_get_gcloud_config", mock_config)
     monkeypatch.setattr(requests, "get", mock_get)
 
     mock_dataset_id = "mock-dataset-id"
@@ -154,6 +162,7 @@ async def test_table_info(monkeypatch, jp_fetch):
 
 async def test_preview(monkeypatch, jp_fetch):
     monkeypatch.setattr(credentials, "get_cached", mock_credentials)
+    monkeypatch.setattr(jupyter_config, "async_get_gcloud_config", mock_config)
     monkeypatch.setattr(requests, "get", mock_get)
 
     mock_dataset_id = "mock-dataset-id"
@@ -214,6 +223,7 @@ async def mock_config(config_field):
 
     monkeypatch.setattr(jupyter_config, "async_get_gcloud_config", mock_config)
     monkeypatch.setattr(credentials, "get_cached", mock_credentials)
+    monkeypatch.setattr(jupyter_config, "async_get_gcloud_config", mock_config)
     monkeypatch.setattr(requests, "post", mock_post)
 
     mock_search_string = "mock-search-string"
diff --git a/dataproc_jupyter_plugin/tests/test_composer.py b/dataproc_jupyter_plugin/tests/test_composer.py
new file mode 100644
index 00000000..d124e22a
--- /dev/null
+++ b/dataproc_jupyter_plugin/tests/test_composer.py
@@ -0,0 +1,60 @@
+# Copyright 2024 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import json
+from unittest.mock import Mock
+
+import requests
+from google.cloud import jupyter_config
+
+from dataproc_jupyter_plugin import credentials
+
+
+async def mock_credentials():
+    return {
+        "project_id": "credentials-project",
+        "project_number": 12345,
+        "region_id": "mock-region",
+        "access_token": "mock-token",
+        "config_error": 0,
+        "login_error": 0,
+    }
+
+
+def mock_get(url, headers):
+    mock_resp = Mock()
+    mock_resp.status_code = 200
+    mock_resp.json.return_value = {
+        "environments": [
+            {"name": "projects/mock-project/locations/mock-location/environments/env1"},
+            {"name": "projects/mock-project/locations/mock-location/environments/env2"},
+        ]
+    }
+    return mock_resp
+
+
+async def mock_config(field_name):
+    return None
+
+
+async def test_list_composer(monkeypatch, jp_fetch):
+    monkeypatch.setattr(credentials, "get_cached", mock_credentials)
+    monkeypatch.setattr(requests, "get", mock_get)
+    monkeypatch.setattr(jupyter_config, "async_get_gcloud_config", mock_config)
+    response = await jp_fetch("dataproc-plugin", "composerList")
+    assert response.code == 200
+    payload = json.loads(response.body)
+    assert len(payload) == 2
+    assert payload[0]["name"] == "env1"
+    assert payload[1]["name"] == "env2"
diff --git a/dataproc_jupyter_plugin/tests/test_dataproc.py b/dataproc_jupyter_plugin/tests/test_dataproc.py
new file mode 100644
index 00000000..133163b2
--- /dev/null
+++ b/dataproc_jupyter_plugin/tests/test_dataproc.py
@@ -0,0 +1,91 @@
+# Copyright 2024 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import json
+from unittest.mock import Mock
+
+import requests
+from google.cloud import jupyter_config
+
+from dataproc_jupyter_plugin import credentials
+
+
+async def mock_credentials():
+    return {
+        "project_id": "credentials-project",
+        "project_number": 12345,
+        "region_id": "mock-region",
+        "access_token": "mock-token",
+        "config_error": 0,
+        "login_error": 0,
+    }
+
+
+def mock_get(api_endpoint, headers=None):
+    response = Mock()
+    response.status_code = 200
+    response.json.return_value = {
+        "api_endpoint": api_endpoint,
+        "headers": headers,
+    }
+    return response
+
+
+async def mock_config(field_name):
+    return None
+
+
+async def test_list_clusters(monkeypatch, jp_fetch):
+    monkeypatch.setattr(credentials, "get_cached", mock_credentials)
+    monkeypatch.setattr(jupyter_config, "async_get_gcloud_config", mock_config)
+    monkeypatch.setattr(requests, "get", mock_get)
+    mock_project_id = "credentials-project"
+    mock_page_token = "mock-page-token"
+    mock_region_id = "mock-region"
+    mock_page_size = "mock_page_size"
+    response = await jp_fetch(
+        "dataproc-plugin",
+        "clusterList",
+        params={"pageSize": mock_page_size, "pageToken": mock_page_token},
+    )
+    assert response.code == 200
+    payload = json.loads(response.body)
+    assert (
+        payload["api_endpoint"]
+        == f"https://dataproc.googleapis.com//v1/projects/credentials-project/regions/{mock_region_id}/clusters?pageSize={mock_page_size}&pageToken={mock_page_token}"
+    )
+    assert payload["headers"]["Authorization"] == f"Bearer mock-token"
+
+
+async def test_list_runtime(monkeypatch, jp_fetch):
+    monkeypatch.setattr(credentials, "get_cached", mock_credentials)
+    monkeypatch.setattr(requests, "get", mock_get)
+    monkeypatch.setattr(jupyter_config, "async_get_gcloud_config", mock_config)
+    mock_project_id = "credentials-project"
+    mock_page_token = "mock-page-token"
+    mock_region_id = "mock-region"
+    mock_page_size = "mock_page_size"
+    mock_service_url = "mock_url"
+    response = await jp_fetch(
+        "dataproc-plugin",
+        "runtimeList",
+        params={"pageSize": mock_page_size, "pageToken": mock_page_token},
+    )
+    assert response.code == 200
+    payload = json.loads(response.body)
+    assert (
+        payload["api_endpoint"]
+        == f"https://dataproc.googleapis.com//v1/projects/{mock_project_id}/locations/{mock_region_id}/sessionTemplates?pageSize=mock_page_size&pageToken={mock_page_token}"
+    )
+    assert payload["headers"]["Authorization"] == f"Bearer mock-token"
diff --git a/dataproc_jupyter_plugin/tests/test_executor.py b/dataproc_jupyter_plugin/tests/test_executor.py
new file mode 100644
index 00000000..e2748904
--- /dev/null
+++ b/dataproc_jupyter_plugin/tests/test_executor.py
@@ -0,0 +1,124 @@
+# Copyright 2024 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import json
+import subprocess
+import unittest
+from unittest.mock import AsyncMock, MagicMock, Mock, patch
+
+import pytest
+import requests
+from google.cloud import jupyter_config
+
+from dataproc_jupyter_plugin import credentials
+from dataproc_jupyter_plugin.services import executor
+
+
+async def mock_credentials():
+    return {
+        "project_id": "credentials-project",
+        "project_number": 12345,
+        "region_id": "mock-region",
+        "access_token": "mock-token",
+        "config_error": 0,
+        "login_error": 0,
+    }
+
+
+def mock_get(api_endpoint, headers=None):
+    response = Mock()
+    response.status_code = 200
+    response.json.return_value = {
+        "api_endpoint": api_endpoint,
+        "headers": headers,
+    }
+    return response
+
+
+async def mock_config(field_name):
+    return None
+
+
+class TestExecuteMethod(unittest.TestCase):
+    def setUp(self):
+        self.instance = executor.Client
+        self.input_data = {
+            "dag_id": "test_dag_id",
+            "name": "test_job_name",
+            "composer_environment_name": "test_env",
+            "input_filename": "test_input_file",
+        }
+
+    @patch("models.DescribeJob")
+    @patch("executor.Client.get_bucket", new_callable=AsyncMock)
+    @patch("executor.Client.check_file_exists")
+    @patch("executor.Client.upload_papermill_to_gcs")
+    @patch("executor.Client.upload_input_file_to_gcs")
+    @patch("executor.Client.prepare_dag")
+    @patch("executor.Client.upload_dag_to_gcs")
+    async def test_execute_success(
+        self,
+        mock_upload_dag_to_gcs,
+        mock_prepare_dag,
+        mock_upload_input_file_to_gcs,
+        mock_upload_papermill_to_gcs,
+        mock_check_file_exists,
+        mock_get_bucket,
+        mock_describe_job,
+    ):
+        mock_job = MagicMock()
+        mock_job.dag_id = "test_dag_id"
+        mock_job.name = "test_job_name"
+        mock_job.composer_environment_name = "test_env"
+        mock_job.input_filename = "test_input_file"
+        mock_describe_job.return_value = mock_job
+
+        mock_get_bucket.return_value = "test_bucket"
+        mock_check_file_exists.return_value = True
+
+        result = await self.instance.execute(self.input_data)
+
+        self.assertEqual(result, {"status": 0})
+
+
+@pytest.mark.parametrize("returncode, expected_result", [(0, 0)])
+async def test_downlaod_output(monkeypatch, returncode, expected_result, jp_fetch):
+    def mock_popen(returncode=0):
+        def _mock_popen(*args, **kwargs):
+            mock_process = Mock()
+            mock_process.communicate.return_value = (b"output", b"")
+            mock_process.returncode = returncode
+            return mock_process
+
+        return _mock_popen
+
+    monkeypatch.setattr(credentials, "get_cached", mock_credentials)
+    monkeypatch.setattr(jupyter_config, "async_get_gcloud_config", mock_config)
+    monkeypatch.setattr(requests, "get", mock_get)
+    monkeypatch.setattr(subprocess, "Popen", mock_popen(returncode))
+    mock_bucket_name = "mock_bucekt"
+    mock_dag_id = "mock_dag_id"
+    mock_dag_run_id = "mock_dag_run_id"
+    response = await jp_fetch(
+        "dataproc-plugin",
+        "downloadOutput",
+        params={
+            "bucket_name": mock_bucket_name,
+            "dag_id": mock_dag_id,
+            "dag_run_id": mock_dag_run_id,
+        },
+    )
+    assert response.code == 200
+    payload = json.loads(response.body)
+    assert payload["status"] == 0
diff --git a/ui-tests-3.6.6/tests/settings_menu.spec.ts b/ui-tests-3.6.6/tests/settings_menu.spec.ts
index cce98b56..224218eb 100644
--- a/ui-tests-3.6.6/tests/settings_menu.spec.ts
+++ b/ui-tests-3.6.6/tests/settings_menu.spec.ts
@@ -23,7 +23,7 @@ test.describe('Settings Menu', () => {
       .getByLabel('main', { exact: true })
       .getByText('Settings')
       .click();
-    await page.getByText('Google BigQuery Settings').click();
+    await page.getByText('Google Dataproc Settings').click();
   });
 
   test('Can change project', async ({ page }) => {
@@ -31,7 +31,7 @@ test.describe('Settings Menu', () => {
       .getByLabel('main', { exact: true })
       .getByText('Settings')
       .click();
-    await page.getByText('Google BigQuery Settings').click();
+    await page.getByText('Google Dataproc Settings').click();
 
     // Assert clearing the Project ID disables the save button.
     await page.getByRole('combobox', { name: 'Project ID' }).click();
diff --git a/ui-tests/tests/settings_menu.spec.ts b/ui-tests/tests/settings_menu.spec.ts
index 10b98ece..21c7327c 100644
--- a/ui-tests/tests/settings_menu.spec.ts
+++ b/ui-tests/tests/settings_menu.spec.ts
@@ -23,7 +23,7 @@ test.describe('Settings Menu', () => {
       .getByLabel('main menu', { exact: true })
       .getByText('Settings')
       .click();
-    await page.getByText('Google BigQuery Settings').click();
+    await page.getByText('Google Dataproc Settings').click();
   });
 
   test('Can change project', async ({ page }) => {
@@ -31,7 +31,7 @@ test.describe('Settings Menu', () => {
       .getByLabel('main menu', { exact: true })
       .getByText('Settings')
       .click();
-    await page.getByText('Google BigQuery Settings').click();
+    await page.getByText('Google Dataproc Settings').click();
 
     // Assert clearing the Project ID disables the save button.
     await page.getByRole('combobox', { name: 'Project ID' }).click();