[superset] branch master updated: fix: Use Celery task ETA for alert/report schedule (#24537)

[jo...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

johnbodley pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/superset.git


The following commit(s) were added to refs/heads/master by this push:
     new e402c94a9f fix: Use Celery task ETA for alert/report schedule (#24537)
e402c94a9f is described below

commit e402c94a9f56766443f8e5fd05f7796fa80355d1
Author: John Bodley <45...@users.noreply.github.com>
AuthorDate: Wed Jun 28 10:53:13 2023 -0700

    fix: Use Celery task ETA for alert/report schedule (#24537)
---
 superset/tasks/scheduler.py                        | 15 ++++-----------
 tests/integration_tests/reports/scheduler_tests.py |  4 ++--
 2 files changed, 6 insertions(+), 13 deletions(-)

diff --git a/superset/tasks/scheduler.py b/superset/tasks/scheduler.py
index 7e0422b001..a84036c340 100644
--- a/superset/tasks/scheduler.py
+++ b/superset/tasks/scheduler.py
@@ -18,7 +18,6 @@ import logging
 
 from celery import Celery
 from celery.exceptions import SoftTimeLimitExceeded
-from dateutil import parser
 
 from superset import app, is_feature_enabled
 from superset.commands.exceptions import CommandException
@@ -64,21 +63,15 @@ def scheduler() -> None:
                         active_schedule.working_timeout
                         + app.config["ALERT_REPORTS_WORKING_SOFT_TIME_OUT_LAG"]
                     )
-                execute.apply_async(
-                    (
-                        active_schedule.id,
-                        schedule,
-                    ),
-                    **async_options,
-                )
+                execute.apply_async((active_schedule.id,), **async_options)
 
 
 @celery_app.task(name="reports.execute", bind=True)
-def execute(self: Celery.task, report_schedule_id: int, scheduled_dttm: str) -> None:
+def execute(self: Celery.task, report_schedule_id: int) -> None:
     task_id = None
     try:
         task_id = execute.request.id
-        scheduled_dttm_ = parser.parse(scheduled_dttm)
+        scheduled_dttm = execute.request.eta
         logger.info(
             "Executing alert/report, task id: %s, scheduled_dttm: %s",
             task_id,
@@ -87,7 +80,7 @@ def execute(self: Celery.task, report_schedule_id: int, scheduled_dttm: str) ->
         AsyncExecuteReportScheduleCommand(
             task_id,
             report_schedule_id,
-            scheduled_dttm_,
+            scheduled_dttm,
         ).run()
     except ReportScheduleUnexpectedError:
         logger.exception(
diff --git a/tests/integration_tests/reports/scheduler_tests.py b/tests/integration_tests/reports/scheduler_tests.py
index 3284ee9772..29dd58273a 100644
--- a/tests/integration_tests/reports/scheduler_tests.py
+++ b/tests/integration_tests/reports/scheduler_tests.py
@@ -171,7 +171,7 @@ def test_execute_task(update_state_mock, command_mock, init_mock, owners):
         init_mock.return_value = None
         command_mock.side_effect = ReportScheduleUnexpectedError("Unexpected error")
         with freeze_time("2020-01-01T09:00:00Z"):
-            execute(report_schedule.id, "2020-01-01T09:00:00Z")
+            execute(report_schedule.id)
             update_state_mock.assert_called_with(state="FAILURE")
 
         db.session.delete(report_schedule)
@@ -199,7 +199,7 @@ def test_execute_task_with_command_exception(
         init_mock.return_value = None
         command_mock.side_effect = CommandException("Unexpected error")
         with freeze_time("2020-01-01T09:00:00Z"):
-            execute(report_schedule.id, "2020-01-01T09:00:00Z")
+            execute(report_schedule.id)
             update_state_mock.assert_called_with(state="FAILURE")
             logger_mock.exception.assert_called_with(
                 "A downstream exception occurred while generating a report: None. Unexpected error",