Repository: flink-extended/dl-on-flink (language: python)

Issue:
[Flink AI Flow] Refractor scheduler config and adjust the position of SchedulerFactory
<!--

*Thank you very much for contributing to flink-ai-extended. To help the community review your issue or contribution in the best possible way，please take a few minutes to fulfill following items.*

-->

### Describe the feature
Current organization of `fink-ai-extended/flink-ai-flow/ai_flow/scheduler` has some problems:
1. The actual position of `Scheduler` class is  `flink-ai-flow/ai_flow/plugin_interface/scheduler_interface.py:`.  Current `scheduler` folder is actually the `SchedulerService`.
2. The. design of `SchedulerServiceConfig` has some isssues:
 - should rename to `SchedulerConfig`
 - should add validation in the constructor method.

### Describe the solution you'd like
<!-- A clear and concise description of what you want to happen -->

### Describe alternatives you've considered
<!-- A clear and concise description of any alternative solutions or features you've considered -->

### Additional context
<!-- Add any other context or screenshots about the feature request here -->



Discussion / hints:

Fix patch:
diff --git a/flink-ai-flow/ai_flow/config_templates/default_aiflow_server.yaml b/flink-ai-flow/ai_flow/config_templates/default_aiflow_server.yaml
index 70d4e0b81..06d0f1922 100644
--- a/flink-ai-flow/ai_flow/config_templates/default_aiflow_server.yaml
+++ b/flink-ai-flow/ai_flow/config_templates/default_aiflow_server.yaml
@@ -48,10 +48,11 @@ db_type: {AIFLOW_DB_TYPE}
 #start_scheduler_service: True
 
 # scheduler config
-scheduler:
-  scheduler_class: ai_flow_plugins.scheduler_plugins.airflow.airflow_scheduler.AirFlowScheduler
-  scheduler_config:
-    # AirFlow dag file deployment directory, i.e., where the airflow dag will be.
-    airflow_deploy_path: {AIFLOW_HOME}/airflow_deploy
-    # Notification service uri used by the scheduler service.
-    notification_service_uri: localhost:50051
+scheduler_service:
+  scheduler:
+    scheduler_class: ai_flow_plugins.scheduler_plugins.airflow.airflow_scheduler.AirFlowScheduler
+    scheduler_config:
+      # AirFlow dag file deployment directory, i.e., where the airflow dag will be.
+      airflow_deploy_path: {AIFLOW_HOME}/airflow_deploy
+      # Notification service uri used by the AirFlowScheduler.
+      notification_service_uri: localhost:50051
diff --git a/flink-ai-flow/ai_flow/endpoint/server/server.py b/flink-ai-flow/ai_flow/endpoint/server/server.py
index 364732f19..9c78ae8ac 100644
--- a/flink-ai-flow/ai_flow/endpoint/server/server.py
+++ b/flink-ai-flow/ai_flow/endpoint/server/server.py
@@ -47,7 +47,7 @@
 from ai_flow.metadata_store.service.service import MetadataService
 from ai_flow.model_center.service.service import ModelCenterService
 from ai_flow.metric.service.metric_service import MetricService
-from ai_flow.scheduler.scheduler_service import SchedulerService, SchedulerServiceConfig
+from ai_flow.scheduler_service.service.service import SchedulerService, SchedulerServiceConfig
 
 sys.path.append(os.path.abspath(os.path.join(os.getcwd(), "../../..")))
 
@@ -113,21 +113,7 @@ def __init__(self, store_uri=None, port=_PORT,
 
     def _add_scheduler_service(self, scheduler_service_config):
         logging.info("start scheduler service.")
-        # check the `scheduler` option of scheduler service config
-        if scheduler_service_config is None:
-            raise Exception(
-                'The `scheduler` option of scheduler service config is not configured. '
-                'Please add the `scheduler` option!')
-        if 'scheduler_class' not in scheduler_service_config:
-            raise Exception(
-                'The `scheduler_class` option of scheduler service config is not configured. '
-                'Please add the `scheduler_class` option under the `scheduler` option!')
-        if 'scheduler_config' not in scheduler_service_config:
-            scheduler_service_config['scheduler_config'] = {}
-        real_config = SchedulerServiceConfig()
-        real_config.set_scheduler_config(scheduler_service_config.get('scheduler_config'))
-        real_config.set_repository(scheduler_service_config.get('repository'))
-        real_config.set_scheduler_class(scheduler_service_config.get('scheduler_class'))
+        real_config = SchedulerServiceConfig(scheduler_service_config)
         self.scheduler_service = SchedulerService(real_config)
         scheduling_service_pb2_grpc.add_SchedulingServiceServicer_to_server(self.scheduler_service,
                                                                             self.server)
diff --git a/flink-ai-flow/ai_flow/endpoint/server/server_config.py b/flink-ai-flow/ai_flow/endpoint/server/server_config.py
index b43dbfeed..905ac6052 100644
--- a/flink-ai-flow/ai_flow/endpoint/server/server_config.py
+++ b/flink-ai-flow/ai_flow/endpoint/server/server_config.py
@@ -99,11 +99,11 @@ def set_notification_uri(self, notification_uri):
     def get_notification_uri(self):
         return self.get('notification_uri')
 
-    def get_scheduler_config(self):
-        return self.get('scheduler')
+    def get_scheduler_service_config(self):
+        return self.get('scheduler_service')
 
-    def set_scheduler_config(self, value):
-        self['scheduler'] = value
+    def set_scheduler_service_config(self, value):
+        self['scheduler_service'] = value
 
     def start_meta_service(self):
         if "start_meta_service" in self and self['start_meta_service'] is False:
diff --git a/flink-ai-flow/ai_flow/endpoint/server/server_runner.py b/flink-ai-flow/ai_flow/endpoint/server/server_runner.py
index c9a144326..b5af42c40 100644
--- a/flink-ai-flow/ai_flow/endpoint/server/server_runner.py
+++ b/flink-ai-flow/ai_flow/endpoint/server/server_runner.py
@@ -74,7 +74,7 @@ def start(self,
             start_model_center_service=self.server_config.start_model_center_service(),
             start_metric_service=self.server_config.start_metric_service(),
             start_scheduler_service=self.server_config.start_scheduler_service(),
-            scheduler_service_config=self.server_config.get_scheduler_config(),
+            scheduler_service_config=self.server_config.get_scheduler_service_config(),
             enabled_ha=self.server_config.get_enable_ha(),
             ha_server_uri=self.server_config.get_server_ip() + ":" + str(self.server_config.get_server_port()),
             ttl_ms=self.server_config.get_ha_ttl_ms())
diff --git a/flink-ai-flow/ai_flow/plugin_interface/blob_manager_interface.py b/flink-ai-flow/ai_flow/plugin_interface/blob_manager_interface.py
index cb86921af..6988ad376 100644
--- a/flink-ai-flow/ai_flow/plugin_interface/blob_manager_interface.py
+++ b/flink-ai-flow/ai_flow/plugin_interface/blob_manager_interface.py
@@ -27,15 +27,19 @@ class BlobConfig(AIFlowConfiguration):
 
     def __init__(self, config: Dict):
         super().__init__()
+        if config is None:
+            raise Exception(
+                'The `{}` option is not configured in the {} option. Please add it!'.format('blob', 'project.yaml'))
         self['blob_manager_class'] = None
-        self['blob_manager_config'] = {}
-        if config is None or 'blob_manager_class' not in config:
+        if config.get('blob_manager_class') is None:
             raise Exception(
                 'The `blob_manager_class` option of blob config is not configured. '
                 'Please add the `blob_manager_class` option under the `blob` option!')
-        self.set_blob_manager_class(config.get('blob_manager_class'))
+        self['blob_manager_class'] = config.get('blob_manager_class')
+        self['blob_manager_config'] = {}
         if config.get('blob_manager_config') is not None:
-            self['blob_manager_config'].update(config.get('blob_manager_config'))
+            self['blob_manager_config'] = config.get('blob_manager_config')
+
 
     def blob_manager_class(self):
         return self.get('blob_manager_class')
@@ -44,8 +48,6 @@ def set_blob_manager_class(self, value):
         self['blob_manager_class'] = value
 
     def blob_manager_config(self):
-        if 'blob_manager_config' not in self:
-            return None
         return self['blob_manager_config']
 
     def set_blob_manager_config(self, value):
diff --git a/flink-ai-flow/ai_flow/plugin_interface/scheduler_interface.py b/flink-ai-flow/ai_flow/plugin_interface/scheduler_interface.py
index b7e6e48bc..e767da61f 100644
--- a/flink-ai-flow/ai_flow/plugin_interface/scheduler_interface.py
+++ b/flink-ai-flow/ai_flow/plugin_interface/scheduler_interface.py
@@ -16,9 +16,43 @@
 # under the License.
 from abc import ABC, abstractmethod
 from typing import Dict, Text, List, Optional
+
 from ai_flow.util import json_utils
 from ai_flow.workflow.workflow import Workflow
 from ai_flow.context.project_context import ProjectContext
+from ai_flow.common.configuration import AIFlowConfiguration
+from ai_flow.common.module_load import import_string
+
+
+class SchedulerConfig(AIFlowConfiguration):
+
+    def __init__(self, config: Dict):
+        super().__init__()
+        if config is None:
+            raise Exception(
+                'The `{}` option is not configured in the {} option. Please add it!'.format('scheduler',
+                                                                                            'scheduler_service'))
+        self['scheduler_class'] = None
+        if config.get('scheduler_class') is None:
+            raise Exception(
+                'The `scheduler_class` option of scheduler config is not configured. '
+                'Please add the `scheduler_class` option under the `scheduler` option!')
+        self['scheduler_class'] = config.get('scheduler_class')
+        self['scheduler_config'] = {}
+        if config.get('scheduler_config') is not None:
+            self['scheduler_config'] = config.get('scheduler_config')
+
+    def scheduler_class(self):
+        return self['scheduler_class']
+
+    def set_scheduler_class(self, value):
+        self['scheduler_class'] = value
+
+    def scheduler_config(self):
+        return self['scheduler_config']
+
+    def set_scheduler_config(self, value):
+        self['scheduler_config'] = value
 
 
 class WorkflowInfo(json_utils.Jsonable):
@@ -355,3 +389,18 @@ def list_job_executions(self, workflow_execution_id: Text) -> List[JobExecutionI
         :return: The job execution information.
         """
         pass
+
+
+class SchedulerFactory(object):
+    """
+    SchedulerFactory creates scheduler based on configuration information.
+    """
+
+    @classmethod
+    def create_scheduler(cls, class_name, config: Dict) -> Scheduler:
+        """
+        :param class_name: The class name of a :py:class:`ai_flow.plugin_interface.scheduler_interface.Scheduler`
+        :param config: The configuration of the scheduler.
+        """
+        class_object = import_string(class_name)
+        return class_object(config)
diff --git a/flink-ai-flow/ai_flow/scheduler/__init__.py b/flink-ai-flow/ai_flow/scheduler_service/__init__.py
similarity index 100%
rename from flink-ai-flow/ai_flow/scheduler/__init__.py
rename to flink-ai-flow/ai_flow/scheduler_service/__init__.py
diff --git a/flink-ai-flow/ai_flow/scheduler/scheduler_factory.py b/flink-ai-flow/ai_flow/scheduler_service/service/__init__.py
similarity index 53%
rename from flink-ai-flow/ai_flow/scheduler/scheduler_factory.py
rename to flink-ai-flow/ai_flow/scheduler_service/service/__init__.py
index 00e05b4d8..d216be4dd 100644
--- a/flink-ai-flow/ai_flow/scheduler/scheduler_factory.py
+++ b/flink-ai-flow/ai_flow/scheduler_service/service/__init__.py
@@ -13,22 +13,4 @@
 # "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
-
-from ai_flow.common.module_load import import_string
-from ai_flow.plugin_interface.scheduler_interface import Scheduler
-
-
-class SchedulerFactory(object):
-    """
-    SchedulerFactory creates scheduler() based on configuration information.
-    """
-    @classmethod
-    def create_scheduler(cls, class_name, config: Dict) -> Scheduler:
-        """
-        :param class_name: The class name of a scheduler(ai_flow.plugin_interface.scheduler_interface.Scheduler)
-        :param config: The configuration of the scheduler.
-        """
-        class_object = import_string(class_name)
-        return class_object(config)
+# under the License.
\ No newline at end of file
diff --git a/flink-ai-flow/ai_flow/scheduler/scheduler_service.py b/flink-ai-flow/ai_flow/scheduler_service/service/service.py
similarity index 94%
rename from flink-ai-flow/ai_flow/scheduler/scheduler_service.py
rename to flink-ai-flow/ai_flow/scheduler_service/service/service.py
index f9471da12..fd6460193 100644
--- a/flink-ai-flow/ai_flow/scheduler/scheduler_service.py
+++ b/flink-ai-flow/ai_flow/scheduler_service/service/service.py
@@ -14,7 +14,7 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-from typing import Text
+from typing import Text, Dict
 import traceback
 
 from ai_flow.common.configuration import AIFlowConfiguration
@@ -35,8 +35,7 @@
      ScheduleJobRequest,
      JobInfoResponse,
      ListJobInfoResponse)
-from ai_flow.scheduler.scheduler_factory import SchedulerFactory
-from ai_flow.plugin_interface.scheduler_interface import Scheduler
+from ai_flow.plugin_interface.scheduler_interface import Scheduler, SchedulerFactory, SchedulerConfig
 from ai_flow.workflow.workflow import Workflow
 from ai_flow.endpoint.server.workflow_proto_utils import workflow_to_proto, workflow_list_to_proto, \
     workflow_execution_to_proto, workflow_execution_list_to_proto, job_to_proto, job_list_to_proto
@@ -44,35 +43,30 @@
 
 class SchedulerServiceConfig(AIFlowConfiguration):
 
-    def __init__(self):
+    def __init__(self, config: Dict):
         super().__init__()
-        self['scheduler_config'] = {}
+        if config is None:
+            raise Exception(
+                'The `{}` option is not configured in the {} option. Please add it!'.format('scheduler_service',
+                                                                                            'aiflow_server.yaml'))
+
+        self['repository'] = '/tmp'
+        if config.get('repository') is not None:
+            self['repository'] = config.get('repository')
+        scheduler_meta = SchedulerConfig(config.get('scheduler'))
+        self['scheduler'] = scheduler_meta
 
     def repository(self):
-        if 'repository' not in self:
-            return '/tmp'
-        else:
-            return self['repository']
+        return self['repository']
 
     def set_repository(self, value):
         self['repository'] = value
 
-    def scheduler_class(self):
-        if self.get('scheduler_class') is not None:
-            return self.get('scheduler_class')
-        else:
-            return None
-
-    def set_scheduler_class(self, value):
-        self['scheduler_class'] = value
-
-    def scheduler_config(self):
-        if 'scheduler_config' not in self:
-            return None
-        return self['scheduler_config']
+    def scheduler(self):
+        return self['scheduler']
 
-    def set_scheduler_config(self, value):
-        self['scheduler_config'] = value
+    def set_scheduler(self, value):
+        self['scheduler'] = value
 
 
 class SchedulerService(SchedulingServiceServicer):
@@ -80,8 +74,8 @@ def __init__(self,
                  scheduler_service_config: SchedulerServiceConfig):
         self._scheduler_service_config = scheduler_service_config
         self._scheduler: Scheduler \
-            = SchedulerFactory.create_scheduler(scheduler_service_config.scheduler_class(),
-                                                scheduler_service_config.scheduler_config())
+            = SchedulerFactory.create_scheduler(scheduler_service_config.scheduler().scheduler_class(),
+                                                scheduler_service_config.scheduler().scheduler_config())
 
     # workflow interface
     def submitWorkflow(self, request, context):
@@ -96,7 +90,7 @@ def submitWorkflow(self, request, context):
             blob_config = BlobConfig(raw_config)
             blob_manager = BlobManagerFactory.create_blob_manager(blob_config.blob_manager_class(),
                                                                   blob_config.blob_manager_config())
-            project_path: Text = blob_manager\
+            project_path: Text = blob_manager \
                 .download_project(workflow_snapshot_id=workflow.workflow_snapshot_id,
                                   remote_path=workflow.project_uri,
                                   local_path=self._scheduler_service_config.repository())
diff --git a/flink-ai-flow/ai_flow_plugins/blob_manager_plugins/local_blob_manager.py b/flink-ai-flow/ai_flow_plugins/blob_manager_plugins/local_blob_manager.py
index e1aea6287..e13a14133 100644
--- a/flink-ai-flow/ai_flow_plugins/blob_manager_plugins/local_blob_manager.py
+++ b/flink-ai-flow/ai_flow_plugins/blob_manager_plugins/local_blob_manager.py
@@ -70,7 +70,7 @@ def download_project(self, workflow_snapshot_id, remote_path: Text, local_path:
         :param local_path: Download file root path.
         :return The local project path.
         """
-        if local_path is not None or self._local_repo is not None:
+        if self._local_repo is not None:
             repo_path = local_path if local_path is not None else self._local_repo
             local_zip_file_name = 'workflow_{}_project'.format(workflow_snapshot_id)
             extract_path = str(Path(repo_path) / local_zip_file_name)
diff --git a/flink-ai-flow/examples/example_utils/master.yaml b/flink-ai-flow/examples/example_utils/master.yaml
index cf470a6e7..e41004d0f 100644
--- a/flink-ai-flow/examples/example_utils/master.yaml
+++ b/flink-ai-flow/examples/example_utils/master.yaml
@@ -21,8 +21,9 @@ server_port: 50051
 db_uri: sqlite:///aiflow.db
 db_type: sql_lite
 start_scheduler_service: True
-scheduler:
-  scheduler_class: ai_flow_plugins.scheduler_plugins.airflow.airflow_scheduler.AirFlowScheduler
-  scheduler_config:
-    airflow_deploy_path: /tmp/airflow
-    notification_service_uri: localhost:50051
+scheduler_service:
+  scheduler:
+    scheduler_class: ai_flow_plugins.scheduler_plugins.airflow.airflow_scheduler.AirFlowScheduler
+    scheduler_config:
+      airflow_deploy_path: /tmp/airflow
+      notification_service_uri: localhost:50051


Test patch:
diff --git a/flink-ai-flow/ai_flow/test/api/ut_workflows/workflows/test_workflow_operation/test_workflow_operation.py b/flink-ai-flow/ai_flow/test/api/ut_workflows/workflows/test_workflow_operation/test_workflow_operation.py
index cc9b64804..6d7f21503 100644
--- a/flink-ai-flow/ai_flow/test/api/ut_workflows/workflows/test_workflow_operation/test_workflow_operation.py
+++ b/flink-ai-flow/ai_flow/test/api/ut_workflows/workflows/test_workflow_operation/test_workflow_operation.py
@@ -22,8 +22,7 @@
 from ai_flow.api.ai_flow_context import init_ai_flow_context
 from ai_flow.context.workflow_config_loader import current_workflow_config
 from ai_flow.api import workflow_operation
-from ai_flow.scheduler.scheduler_service import SchedulerServiceConfig
-from ai_flow.test.api.mock_plugins import MockJobFactory
+from ai_flow.scheduler_service.service.service import SchedulerServiceConfig
 
 _SQLITE_DB_FILE = 'aiflow.db'
 _SQLITE_DB_URI = '%s%s' % ('sqlite:///', _SQLITE_DB_FILE)
@@ -39,8 +38,12 @@ def setUpClass(cls):
 
         if os.path.exists(_SQLITE_DB_FILE):
             os.remove(_SQLITE_DB_FILE)
-        config = SchedulerServiceConfig()
-        config.set_scheduler_class(SCHEDULER_CLASS)
+        raw_config = {
+            'scheduler': {
+                'scheduler_class': SCHEDULER_CLASS,
+            }
+        }
+        config = SchedulerServiceConfig(raw_config)
         cls.server = AIFlowServer(store_uri=_SQLITE_DB_URI, port=_PORT,
                                   start_default_notification=False,
                                   start_meta_service=True,
diff --git a/flink-ai-flow/ai_flow/test/endpoint/server/master_config.yaml b/flink-ai-flow/ai_flow/test/endpoint/server/master_config.yaml
index 0f4d7250f..a381b487b 100644
--- a/flink-ai-flow/ai_flow/test/endpoint/server/master_config.yaml
+++ b/flink-ai-flow/ai_flow/test/endpoint/server/master_config.yaml
@@ -21,8 +21,9 @@ server_port: 50051
 db_uri: sqlite:///aiflow.db
 db_type: sql_lite
 start_scheduler_service: True
-scheduler:
-  scheduler_class: ai_flow.test.endpoint.server.test_server_runner.MockScheduler
+scheduler_service:
   repository: /tmp/repo
-  scheduler_config:
-    notification_service_uri: localhost:9999
\ No newline at end of file
+  scheduler:
+    scheduler_class: ai_flow.test.endpoint.server.test_server_runner.MockScheduler
+    scheduler_config:
+      notification_service_uri: localhost:9999
\ No newline at end of file
diff --git a/flink-ai-flow/ai_flow/test/endpoint/server/test_server_config.py b/flink-ai-flow/ai_flow/test/endpoint/server/test_server_config.py
index 1ed7c19ea..8ef892103 100644
--- a/flink-ai-flow/ai_flow/test/endpoint/server/test_server_config.py
+++ b/flink-ai-flow/ai_flow/test/endpoint/server/test_server_config.py
@@ -32,7 +32,7 @@ def test_load_master_configuration(self):
         config = AIFlowServerConfig()
         config.load_from_file(os.path.dirname(__file__) + '/master_config.yaml')
         self.assertEqual('sql_lite', config.get_db_type())
-        self.assertEqual('/tmp/repo', config.get_scheduler_config()['repository'])
+        self.assertEqual('/tmp/repo', config.get_scheduler_service_config()['repository'])
         self.assertEqual(True, config.start_scheduler_service())
 
 
diff --git a/flink-ai-flow/ai_flow/test/scheduler/test_scheduler_factory.py b/flink-ai-flow/ai_flow/test/scheduler/test_scheduler_factory.py
index 5e7ca91d7..bc7a7af00 100644
--- a/flink-ai-flow/ai_flow/test/scheduler/test_scheduler_factory.py
+++ b/flink-ai-flow/ai_flow/test/scheduler/test_scheduler_factory.py
@@ -18,8 +18,7 @@
 from typing import Text, List, Dict
 
 from ai_flow.context.project_context import ProjectContext
-from ai_flow.scheduler.scheduler_factory import SchedulerFactory
-from ai_flow.plugin_interface.scheduler_interface import Scheduler
+from ai_flow.plugin_interface.scheduler_interface import SchedulerFactory, Scheduler
 from ai_flow.workflow.workflow import Workflow
 from ai_flow.plugin_interface.scheduler_interface import JobExecutionInfo, WorkflowExecutionInfo, WorkflowInfo
 
diff --git a/flink-ai-flow/ai_flow/test/scheduler/test_scheduler_service.py b/flink-ai-flow/ai_flow/test/scheduler/test_scheduler_service.py
index 21251582a..7e77097c2 100644
--- a/flink-ai-flow/ai_flow/test/scheduler/test_scheduler_service.py
+++ b/flink-ai-flow/ai_flow/test/scheduler/test_scheduler_service.py
@@ -18,9 +18,9 @@
 import os
 import unittest
 
-from typing import Text, List, Dict
+from typing import Text, List
 
-from ai_flow.scheduler.scheduler_service import SchedulerServiceConfig
+from ai_flow.scheduler_service.service.service import SchedulerServiceConfig
 from ai_flow.workflow.status import Status
 from ai_flow.protobuf.message_pb2 import StateProto
 
@@ -85,9 +85,12 @@ def list_workflow_executions(self, project_name: Text, workflow_name: Text) -> L
 
 class TestSchedulerService(unittest.TestCase):
     def setUp(self):
-        config = SchedulerServiceConfig()
-        config.set_scheduler_class(SCHEDULER_CLASS)
-        config.set_scheduler_config({})
+        raw_config = {
+            'scheduler': {
+                'scheduler_class': SCHEDULER_CLASS,
+            }
+        }
+        config = SchedulerServiceConfig(raw_config)
         if os.path.exists(_SQLITE_DB_FILE):
             os.remove(_SQLITE_DB_FILE)
         self.server = AIFlowServer(store_uri=_SQLITE_DB_URI, port=_PORT,
diff --git a/flink-ai-flow/ai_flow_plugins/tests/it_workflows/master.yaml b/flink-ai-flow/ai_flow_plugins/tests/it_workflows/master.yaml
index cf470a6e7..b902a822f 100644
--- a/flink-ai-flow/ai_flow_plugins/tests/it_workflows/master.yaml
+++ b/flink-ai-flow/ai_flow_plugins/tests/it_workflows/master.yaml
@@ -21,8 +21,10 @@ server_port: 50051
 db_uri: sqlite:///aiflow.db
 db_type: sql_lite
 start_scheduler_service: True
-scheduler:
-  scheduler_class: ai_flow_plugins.scheduler_plugins.airflow.airflow_scheduler.AirFlowScheduler
-  scheduler_config:
-    airflow_deploy_path: /tmp/airflow
-    notification_service_uri: localhost:50051
+
+scheduler_service:
+  scheduler:
+    scheduler_class: ai_flow_plugins.scheduler_plugins.airflow.airflow_scheduler.AirFlowScheduler
+    scheduler_config:
+      airflow_deploy_path: /tmp/airflow
+      notification_service_uri: localhost:50051
diff --git a/flink-ai-flow/ai_flow_plugins/tests/job_plugins/ut_workflows/master.yaml b/flink-ai-flow/ai_flow_plugins/tests/job_plugins/ut_workflows/master.yaml
index c533d580d..2b51946ed 100644
--- a/flink-ai-flow/ai_flow_plugins/tests/job_plugins/ut_workflows/master.yaml
+++ b/flink-ai-flow/ai_flow_plugins/tests/job_plugins/ut_workflows/master.yaml
@@ -21,5 +21,6 @@ server_port: 50051
 db_uri: sqlite:///aiflow.db
 db_type: sql_lite
 start_scheduler_service: True
-scheduler:
-  scheduler_class: ai_flow_plugins.tests.job_plugins.single_job_scheduler.SingleJobScheduler
+scheduler_service:
+  scheduler:
+    scheduler_class: ai_flow_plugins.tests.job_plugins.single_job_scheduler.SingleJobScheduler
diff --git a/flink-ai-flow/ai_flow_plugins/tests/scheduler_plugins/airflow/ut_workflows/master.yaml b/flink-ai-flow/ai_flow_plugins/tests/scheduler_plugins/airflow/ut_workflows/master.yaml
index b8f925bae..9a7d77640 100644
--- a/flink-ai-flow/ai_flow_plugins/tests/scheduler_plugins/airflow/ut_workflows/master.yaml
+++ b/flink-ai-flow/ai_flow_plugins/tests/scheduler_plugins/airflow/ut_workflows/master.yaml
@@ -21,5 +21,7 @@ server_port: 50051
 db_uri: sqlite:///aiflow.db
 db_type: sql_lite
 start_scheduler_service: True
-scheduler:
-  scheduler_class: ai_flow_plugins.tests.scheduler_plugins.airflow.mock_scheduler_plugin.MockScheduler
+
+scheduler_service:
+  scheduler:
+    scheduler_class: ai_flow_plugins.tests.scheduler_plugins.airflow.mock_scheduler_plugin.MockScheduler