diff --git a/superset/tasks/scheduler.py b/superset/tasks/scheduler.py
index eacc2ec49b1..4b1debea4b1 100644
--- a/superset/tasks/scheduler.py
+++ b/superset/tasks/scheduler.py
@@ -19,7 +19,7 @@ import logging
 from celery.exceptions import SoftTimeLimitExceeded
 from dateutil import parser
 
-from superset import app
+from superset import app, is_feature_enabled
 from superset.commands.exceptions import CommandException
 from superset.extensions import celery_app
 from superset.reports.commands.exceptions import ReportScheduleUnexpectedError
@@ -37,6 +37,8 @@ def scheduler() -> None:
     """
     Celery beat main scheduler for reports
     """
+    if not is_feature_enabled("ALERT_REPORTS"):
+        return
     with session_scope(nullpool=True) as session:
         active_schedules = ReportScheduleDAO.find_active(session)
         for active_schedule in active_schedules:
diff --git a/tests/integration_tests/reports/scheduler_tests.py b/tests/integration_tests/reports/scheduler_tests.py
index 73783f49676..77894a6d2ef 100644
--- a/tests/integration_tests/reports/scheduler_tests.py
+++ b/tests/integration_tests/reports/scheduler_tests.py
@@ -115,3 +115,25 @@ def test_scheduler_celery_no_timeout_utc(execute_mock):
         db.session.delete(report_schedule)
         db.session.commit()
         app.config["ALERT_REPORTS_WORKING_TIME_OUT_KILL"] = True
+
+
+@patch("superset.tasks.scheduler.is_feature_enabled")
+@patch("superset.tasks.scheduler.execute.apply_async")
+def test_scheduler_feature_flag_off(execute_mock, is_feature_enabled):
+    """
+    Reports scheduler: Test scheduler with feature flag off
+    """
+    with app.app_context():
+        is_feature_enabled.return_value = False
+        report_schedule = insert_report_schedule(
+            type=ReportScheduleType.ALERT,
+            name="report",
+            crontab="0 9 * * *",
+            timezone="UTC",
+        )
+
+        with freeze_time("2020-01-01T09:00:00Z"):
+            scheduler()
+            execute_mock.assert_not_called()
+        db.session.delete(report_schedule)
+        db.session.commit()