manager now fails on nested jobs hitting max attempts

Author: Phlair

airbyte-integrations/connectors/source-facebook-marketing/source_facebook_marketing/streams/async_job.py

@@ -58,9 +58,6 @@ class Status(str, Enum):
 class AsyncJob(ABC):
     """Abstract AsyncJob base class"""
 
-    # max attempts for a job before errroring out
-    max_attempts: int = 10  # TODO: verify a sane number for this
-
     def __init__(self, api: FacebookAdsApi, interval: pendulum.Period):
         """Init generic async job
 
@@ -163,11 +160,9 @@ def split_job(self) -> List["AsyncJob"]:
             if job.failed:
                 try:
                     new_jobs.extend(job.split_job())
-                except RuntimeError as split_limit_error:
+                except ValueError as split_limit_error:
                     logger.error(split_limit_error)
-                    if job.attempt_number > job.max_attempts:
-                        raise RuntimeError(f"{job} at smallest split size and still failing after {job.max_attempts} retries.")
-                    logger.info(f'can\'t split "{job}" any smaller, attempting to restart instead.')
+                    logger.info(f'can\'t split "{job}" any smaller, attempting to retry the job.')
                     job.restart()
                     new_jobs.append(job)
             else:
@@ -213,7 +208,7 @@ def split_job(self) -> List["AsyncJob"]:
             return self._split_by_edge_class(AdSet)
         elif isinstance(self._edge_object, AdSet):
             return self._split_by_edge_class(Ad)
-        raise RuntimeError("The job is already splitted to the smallest size.")
+        raise ValueError("The job is already splitted to the smallest size.")
 
     def _split_by_edge_class(self, edge_class: Union[Type[Campaign], Type[AdSet], Type[Ad]]) -> List[AsyncJob]:
         """Split insight job by creating insight jobs from lower edge object, i.e.

airbyte-integrations/connectors/source-facebook-marketing/source_facebook_marketing/streams/async_job_manager.py

@@ -94,6 +94,12 @@ def _check_jobs_status_and_restart(self) -> List[AsyncJob]:
         self._wait_throttle_limit_down()
         for job in self._running_jobs:
             if job.failed:
+                if isinstance(job, ParentAsyncJob):
+                    # if this job is a ParentAsyncJob, it holds X number of jobs
+                    # we want to check that none of these nested jobs have exceeded MAX_NUMBER_OF_ATTEMPTS
+                    for nested_job in job._jobs:
+                        if nested_job.attempt_number >= self.MAX_NUMBER_OF_ATTEMPTS:
+                            raise JobException(f"{nested_job}: failed more than {self.MAX_NUMBER_OF_ATTEMPTS} times. Terminating...")
                 if job.attempt_number >= self.MAX_NUMBER_OF_ATTEMPTS:
                     raise JobException(f"{job}: failed more than {self.MAX_NUMBER_OF_ATTEMPTS} times. Terminating...")
                 elif job.attempt_number == 2:

airbyte-integrations/connectors/source-facebook-marketing/unit_tests/test_async_job.py

@@ -336,7 +336,7 @@ def test_split_job_smallest(self, mocker, api):
         params = {"time_increment": 1, "breakdowns": []}
         job = InsightAsyncJob(api=api, edge_object=Ad(1), interval=interval, params=params)
 
-        with pytest.raises(RuntimeError, match="The job is already splitted to the smallest size."):
+        with pytest.raises(ValueError, match="The job is already splitted to the smallest size."):
             job.split_job()
 
 
@@ -416,21 +416,18 @@ def test_split_job(self, parent_job, grouped_jobs, mocker):
                 job.split_job.assert_not_called()
 
     def test_split_job_smallest(self, parent_job, grouped_jobs):
-        grouped_jobs[0].max_attempts = InsightAsyncJob.max_attempts
         grouped_jobs[0].failed = True
-        grouped_jobs[0].split_job.side_effect = RuntimeError("Mocking smallest size")
+        grouped_jobs[0].split_job.side_effect = ValueError("Mocking smallest size")
 
+        # arbitrarily testing this X times, the max attempts is handled by async_job_manager rather than the job itself.
         count = 0
-        while count < InsightAsyncJob.max_attempts:
+        while count < 10:
             split_jobs = parent_job.split_job()
             assert len(split_jobs) == len(
                 grouped_jobs
             ), "attempted to split job at smallest size so should just restart job meaning same no. of jobs"
             grouped_jobs[0].attempt_number += 1
             count += 1
 
-        with pytest.raises(RuntimeError):  # now that we've hit max_attempts, we should error out
-            parent_job.split_job()
-
     def test_str(self, parent_job, grouped_jobs):
         assert str(parent_job) == f"ParentAsyncJob({grouped_jobs[0]} ... {len(grouped_jobs) - 1} jobs more)"

airbyte-integrations/connectors/source-facebook-marketing/unit_tests/test_async_job_manager.py

@@ -151,3 +151,26 @@ def update_job_behaviour():
 
         with pytest.raises(JobException, match=f"{jobs[1]}: failed more than {InsightAsyncJobManager.MAX_NUMBER_OF_ATTEMPTS} times."):
             next(manager.completed_jobs(), None)
+
+    def test_nested_job_failed_too_many_times(self, api, mocker, time_mock, update_job_mock):
+        """Manager should fail when a nested job within a ParentAsyncJob failed too many times"""
+
+        def update_job_behaviour():
+            jobs[1].failed = True
+            sub_jobs[1].failed = True
+            sub_jobs[1].attempt_number = InsightAsyncJobManager.MAX_NUMBER_OF_ATTEMPTS
+            yield from range(10)
+
+        update_job_mock.side_effect = update_job_behaviour()
+        sub_jobs = [
+            mocker.Mock(spec=InsightAsyncJob, attempt_number=1, failed=False, completed=True),
+            mocker.Mock(spec=InsightAsyncJob, attempt_number=1, failed=False, completed=False),
+        ]
+        jobs = [
+            mocker.Mock(spec=InsightAsyncJob, attempt_number=1, failed=False, completed=True),
+            mocker.Mock(spec=ParentAsyncJob, _jobs=sub_jobs, attempt_number=1, failed=False, completed=False),
+        ]
+        manager = InsightAsyncJobManager(api=api, jobs=jobs)
+
+        with pytest.raises(JobException):
+            next(manager.completed_jobs(), None)