Source Facebook Marketing: Fix fail when async job takes too long (#7744)

* Update AdsInsights wait_for_job retry logics

* Update wait_for_job restart mapping

* Fix review comments

* Bump docker version

Author: htrueman

airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/e7778cfc-e97c-4458-9ecb-b4f2bba8946c.json

@@ -2,7 +2,7 @@
   "sourceDefinitionId": "e7778cfc-e97c-4458-9ecb-b4f2bba8946c",
   "name": "Facebook Marketing",
   "dockerRepository": "airbyte/source-facebook-marketing",
-  "dockerImageTag": "0.2.22",
+  "dockerImageTag": "0.2.24",
   "documentationUrl": "https://docs.airbyte.io/integrations/sources/facebook-marketing",
   "icon": "facebook.svg"
 }

airbyte-config/init/src/main/resources/seed/source_definitions.yaml

@@ -137,7 +137,7 @@
 - name: Facebook Marketing
   sourceDefinitionId: e7778cfc-e97c-4458-9ecb-b4f2bba8946c
   dockerRepository: airbyte/source-facebook-marketing
-  dockerImageTag: 0.2.22
+  dockerImageTag: 0.2.24
   documentationUrl: https://docs.airbyte.io/integrations/sources/facebook-marketing
   icon: facebook.svg
   sourceType: api

airbyte-integrations/connectors/source-facebook-marketing/Dockerfile

@@ -12,5 +12,5 @@ RUN pip install .
 ENV AIRBYTE_ENTRYPOINT "python /airbyte/integration_code/main.py"
 ENTRYPOINT ["python", "/airbyte/integration_code/main.py"]
 
-LABEL io.airbyte.version=0.2.23
+LABEL io.airbyte.version=0.2.24
 LABEL io.airbyte.name=airbyte/source-facebook-marketing

airbyte-integrations/connectors/source-facebook-marketing/source_facebook_marketing/common.py

@@ -23,6 +23,10 @@ class FacebookAPIException(Exception):
     """General class for all API errors"""
 
 
+class JobException(Exception):
+    """Job failed after FB exception"""
+
+
 class JobTimeoutException(Exception):
     """Scheduled job timed out"""
 

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

@@ -5,7 +5,7 @@
 import time
 import urllib.parse as urlparse
 from abc import ABC
-from collections import deque
+from collections import defaultdict, deque
 from datetime import datetime
 from typing import Any, Iterable, Iterator, List, Mapping, MutableMapping, Optional, Sequence
 
@@ -23,9 +23,10 @@
 from facebook_business.exceptions import FacebookRequestError
 from source_facebook_marketing.api import API
 
-from .common import FacebookAPIException, JobTimeoutException, batch, deep_merge, retry_pattern
+from .common import FacebookAPIException, JobException, JobTimeoutException, batch, deep_merge, retry_pattern
 
 backoff_policy = retry_pattern(backoff.expo, FacebookRequestError, max_tries=5, factor=5)
+MAX_JOB_RESTART_TIMES = 6
 
 
 def remove_params_from_url(url: str, params: List[str]) -> str:
@@ -291,7 +292,7 @@ class AdsInsights(FBMarketingIncrementalStream):
     time_increment = 1
 
     running_jobs = deque()
-    times_job_restarted = {}
+    times_job_restarted = defaultdict(int)
 
     breakdowns = []
 
@@ -369,31 +370,51 @@ def wait_for_job(self, job, stream_state: Mapping[str, Any] = None) -> AdReportR
 
             if job["async_status"] == "Job Completed":
                 return job
-            elif job["async_status"] in ["Job Failed", "Job Skipped"]:
+            else:
                 time_range = (job["date_start"], job["date_stop"])
-                if self.times_job_restarted.get(time_range, 0) < 6:
+                if self.times_job_restarted[time_range] < MAX_JOB_RESTART_TIMES:
                     params = deep_merge(
                         {"time_range": {"since": job["date_start"], "until": job["date_stop"]}},
                         self.request_params(stream_state=stream_state),
                     )
                     restart_job = self._create_insights_job(params)
                     self.running_jobs.append(restart_job)
                     self.times_job_restarted[time_range] += 1
-                elif job["async_status"] == "Job Failed":
-                    raise JobTimeoutException(f"AdReportRun {job} failed after {runtime.in_seconds()} seconds.")
-                elif job["async_status"] == "Job Skipped":
-                    raise JobTimeoutException(f"AdReportRun {job} skipped after {runtime.in_seconds()} seconds.")
-
-            if runtime > self.MAX_WAIT_TO_START and job_progress_pct == 0:
-                raise JobTimeoutException(
-                    f"AdReportRun {job} did not start after {runtime.in_seconds()} seconds."
-                    f" This is an intermittent error which may be fixed by retrying the job. Aborting."
-                )
-            elif runtime > self.MAX_WAIT_TO_FINISH:
-                raise JobTimeoutException(
-                    f"AdReportRun {job} did not finish after {runtime.in_seconds()} seconds."
-                    f" This is an intermittent error which may be fixed by retrying the job. Aborting."
-                )
+                else:
+                    error_msg = """
+                        AdReportRun job with id={id} {error_type}.
+                        Details: report_run_id={report_run_id},
+                        date_start={date_start}, date_stop={date_stop}
+                    """
+                    if job["async_status"] == "Job Failed":
+                        raise JobException(
+                            error_msg.format(
+                                error_type="failed",
+                                report_run_id=job["report_run_id"],
+                                date_start=job["date_start"],
+                                date_stop=job["date_stop"],
+                            )
+                        )
+                    elif job["async_status"] == "Job Skipped":
+                        raise JobException(
+                            error_msg.format(
+                                error_type="failed",
+                                report_run_id=job["report_run_id"],
+                                date_start=job["date_start"],
+                                date_stop=job["date_stop"],
+                            )
+                        )
+
+                    if runtime > self.MAX_WAIT_TO_START and job_progress_pct == 0:
+                        raise JobTimeoutException(
+                            f"AdReportRun {job} did not start after {runtime.in_seconds()} seconds."
+                            f" This is an intermittent error which may be fixed by retrying the job. Aborting."
+                        )
+                    elif runtime > self.MAX_WAIT_TO_FINISH:
+                        raise JobTimeoutException(
+                            f"AdReportRun {job} did not finish after {runtime.in_seconds()} seconds."
+                            f" This is an intermittent error which may be fixed by retrying the job. Aborting."
+                        )
             self.logger.info(f"Sleeping {sleep_seconds} seconds while waiting for AdReportRun: {job_id} to complete")
             time.sleep(sleep_seconds)
             if sleep_seconds < self.MAX_ASYNC_SLEEP.in_seconds():

docs/integrations/sources/facebook-marketing.md

@@ -96,6 +96,7 @@ As a summary, custom insights allows to replicate only some fields, resulting in
 
 | Version | Date | Pull Request | Subject |
 | :--- | :--- | :--- | :--- |
+| 0.2.24  | 2021-11-09 | [7744](https://github.com/airbytehq/airbyte/pull/7744) | Fix fail when async job takes too long |
 | 0.2.23  | 2021-11-08 | [7734](https://github.com/airbytehq/airbyte/pull/7734) | Resolve $ref field for discover schema |
 | 0.2.22  | 2021-11-05 | [7605](https://github.com/airbytehq/airbyte/pull/7605) | Add job retry logics to AdsInsights stream |
 | 0.2.21  | 2021-10-05 | [4864](https://github.com/airbytehq/airbyte/pull/4864) | Update insights streams with custom entries for fields, breakdowns and action_breakdowns |