Source Facebook Marketing: Attempt to retry failing jobs that are already split to minimum size

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

@@ -58,6 +58,9 @@ 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
 
@@ -158,7 +161,15 @@ def split_job(self) -> List["AsyncJob"]:
         new_jobs = []
         for job in self._jobs:
             if job.failed:
-                new_jobs.extend(job.split_job())
+                try:
+                    new_jobs.extend(job.split_job())
+                except RuntimeError 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.')
+                    job.restart()
+                    new_jobs.append(job)
             else:
                 new_jobs.append(job)
         return new_jobs

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

@@ -415,5 +415,22 @@ def test_split_job(self, parent_job, grouped_jobs, mocker):
             else:
                 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")
+
+        count = 0
+        while count < InsightAsyncJob.max_attempts:
+            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)"