From f6295edfcbe5c8f075ac3fa8603a4a53a0c303b3 Mon Sep 17 00:00:00 2001 From: Matt Bertrand Date: Thu, 26 Sep 2024 08:17:26 -0400 Subject: [PATCH 1/2] Add task_reject_on_worker_lost=True to finish_recreate_index, use return instead of raise for replaced tasks (#1608) --- learning_resources/tasks.py | 14 +++++++------- learning_resources_search/tasks.py | 12 ++++++++---- 2 files changed, 15 insertions(+), 11 deletions(-) diff --git a/learning_resources/tasks.py b/learning_resources/tasks.py index 0fd1527a3e..0cd857bcfe 100644 --- a/learning_resources/tasks.py +++ b/learning_resources/tasks.py @@ -170,7 +170,7 @@ def get_content_tasks( @app.task(bind=True) def import_all_mit_edx_files(self, chunk_size=None): """Ingest MIT edX files from an S3 bucket""" - raise self.replace( + return self.replace( get_content_tasks( ETLSource.mit_edx.name, chunk_size=chunk_size, @@ -182,7 +182,7 @@ def import_all_mit_edx_files(self, chunk_size=None): @app.task(bind=True) def import_all_oll_files(self, chunk_size=None): """Ingest MIT edX files from an S3 bucket""" - raise self.replace( + return self.replace( get_content_tasks( ETLSource.oll.name, chunk_size=chunk_size, @@ -195,7 +195,7 @@ def import_all_oll_files(self, chunk_size=None): @app.task(bind=True) def import_all_mitxonline_files(self, chunk_size=None): """Ingest MITx Online files from an S3 bucket""" - raise self.replace( + return self.replace( get_content_tasks( ETLSource.mitxonline.name, chunk_size=chunk_size, @@ -207,7 +207,7 @@ def import_all_mitxonline_files(self, chunk_size=None): def import_all_xpro_files(self, chunk_size=None): """Ingest xPRO OLX files from an S3 bucket""" - raise self.replace( + return self.replace( get_content_tasks( ETLSource.xpro.name, chunk_size=chunk_size, @@ -274,7 +274,7 @@ def get_ocw_data( # noqa: PLR0913 and settings.OCW_LIVE_BUCKET ): log.warning("Required settings missing for get_ocw_data") - return + return None # get all the courses prefixes we care about raw_data_bucket = boto3.resource( @@ -300,7 +300,7 @@ def get_ocw_data( # noqa: PLR0913 if len(ocw_courses) == 0: log.info("No courses matching url substring") - return + return None log.info("Backpopulating %d OCW courses...", len(ocw_courses)) @@ -317,7 +317,7 @@ def get_ocw_data( # noqa: PLR0913 ) ] ) - raise self.replace(ocw_tasks) + return self.replace(ocw_tasks) @app.task diff --git a/learning_resources_search/tasks.py b/learning_resources_search/tasks.py index bf1b3fe598..b3a07cf71c 100644 --- a/learning_resources_search/tasks.py +++ b/learning_resources_search/tasks.py @@ -272,7 +272,7 @@ def send_subscription_emails(self, subscription_type, period="daily"): ) ] ) - raise self.replace(email_tasks) + return self.replace(email_tasks) @app.task(autoretry_for=(RetryError,), retry_backoff=True, rate_limit="600/m") @@ -625,7 +625,7 @@ def start_recreate_index(self, indexes, remove_existing_reindexing_tags): # Use self.replace so that code waiting on this task will also wait on the indexing # and finish tasks - raise self.replace( + return self.replace( celery.chain(index_tasks, finish_recreate_index.s(new_backing_indices)) ) @@ -678,7 +678,7 @@ def start_update_index(self, indexes, etl_source): error = "start_update_index threw an error" log.exception(error) return [error] - raise self.replace(celery.chain(index_tasks, finish_update_index.s())) + return self.replace(celery.chain(index_tasks, finish_update_index.s())) def get_update_resource_files_tasks(blocklisted_ids, etl_source): @@ -841,7 +841,11 @@ def get_update_learning_resource_tasks(resource_type): @app.task( - acks_late=True, autoretry_for=(RetryError,), retry_backoff=True, rate_limit="600/m" + acks_late=True, + reject_on_worker_lost=True, + autoretry_for=(RetryError,), + retry_backoff=True, + rate_limit="600/m", ) def finish_recreate_index(results, backing_indices): """ From 2c84383629c6ca0f412af6a7933cf5f3521b75bc Mon Sep 17 00:00:00 2001 From: Doof Date: Thu, 26 Sep 2024 16:44:29 +0000 Subject: [PATCH 2/2] Release 0.19.6 --- RELEASE.rst | 5 +++++ main/settings.py | 2 +- 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/RELEASE.rst b/RELEASE.rst index 4f5864c9ce..b3338b2eaf 100644 --- a/RELEASE.rst +++ b/RELEASE.rst @@ -1,6 +1,11 @@ Release Notes ============= +Version 0.19.6 +-------------- + +- Add task_reject_on_worker_lost=True to finish_recreate_index, use return instead of raise for replaced tasks (#1608) + Version 0.19.5 (Released September 26, 2024) -------------- diff --git a/main/settings.py b/main/settings.py index 4547c9910c..1373918998 100644 --- a/main/settings.py +++ b/main/settings.py @@ -33,7 +33,7 @@ from main.settings_pluggy import * # noqa: F403 from openapi.settings_spectacular import open_spectacular_settings -VERSION = "0.19.5" +VERSION = "0.19.6" log = logging.getLogger()