🐛 Source Facebook Marketing: Fix backoff trigger with ConnectionError (#10348)

* Add AdAccount and Images stream implementation

* Update PR number

* Updated docker version

* Updated to linter

* Update to review

* Add comment to AdAccount read_records method

* Bumped version in seed, definitions and specs files

* Fix backoff errors code list

* Deleted empty file

* Updated to linter

* Updated PR number

* Added connection_reset_error to backoff

* Fix unittest

* Fix integration tests

* Bumped seed version

Author: lazebnyi

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

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

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

@@ -1648,7 +1648,7 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-facebook-marketing:0.2.34"
+- dockerImage: "airbyte/source-facebook-marketing:0.2.35"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/sources/facebook-marketing"
     changelogUrl: "https://docs.airbyte.io/integrations/sources/facebook-marketing"

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.34
+LABEL io.airbyte.version=0.2.35
 LABEL io.airbyte.name=airbyte/source-facebook-marketing

airbyte-integrations/connectors/source-facebook-marketing/integration_tests/test_streams.py

@@ -33,7 +33,7 @@ def state_with_include_deleted_fixture(state):
 
 @pytest.fixture(scope="session", name="configured_catalog")
 def configured_catalog_fixture():
-    return ConfiguredAirbyteCatalog.parse_file("integration_tests/configured_catalog.json")
+    return ConfiguredAirbyteCatalog.parse_file("integration_tests/configured_catalog_without_insights.json")
 
 
 class TestFacebookMarketingSource:

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

@@ -16,6 +16,7 @@
 FACEBOOK_RATE_LIMIT_ERROR_CODES = (4, 17, 32, 613, 80000, 80001, 80002, 80003, 80004, 80005, 80006, 80008)
 FACEBOOK_BATCH_ERROR_CODE = 960
 FACEBOOK_UNKNOWN_ERROR_CODE = 99
+FACEBOOK_CONNECTION_RESET_ERROR_CODE = 104
 DEFAULT_SLEEP_INTERVAL = pendulum.duration(minutes=1)
 MAX_BATCH_SIZE = 50
 
@@ -37,7 +38,8 @@ def should_retry_api_error(exc):
             call_rate_limit_error = exc.api_error_code() in FACEBOOK_RATE_LIMIT_ERROR_CODES
             batch_timeout_error = exc.http_status() == http.client.BAD_REQUEST and exc.api_error_code() == FACEBOOK_BATCH_ERROR_CODE
             unknown_error = exc.api_error_subcode() == FACEBOOK_UNKNOWN_ERROR_CODE
-            return any((exc.api_transient_error(), unknown_error, call_rate_limit_error, batch_timeout_error))
+            connection_reset_error = exc.api_error_code() == FACEBOOK_CONNECTION_RESET_ERROR_CODE
+            return any((exc.api_transient_error(), unknown_error, call_rate_limit_error, batch_timeout_error, connection_reset_error))
         return True
 
     return backoff.on_exception(

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

@@ -11,7 +11,7 @@
 from facebook_business import FacebookAdsApi, FacebookSession
 from facebook_business.exceptions import FacebookRequestError
 from source_facebook_marketing.api import API
-from source_facebook_marketing.streams import AdCreatives, Campaigns
+from source_facebook_marketing.streams import AdCreatives, AdsInsights, Campaigns
 
 FB_API_VERSION = FacebookAdsApi.API_VERSION
 
@@ -159,3 +159,14 @@ def test_server_error(self, requests_mock, api, account_id):
         with pytest.raises(FacebookRequestError):
             stream = Campaigns(api=api, start_date=datetime.now(), end_date=datetime.now(), include_deleted=False)
             list(stream.read_records(sync_mode=SyncMode.full_refresh, stream_state={}))
+
+    def test_connection_reset_error(self, requests_mock, api, account_id):
+        """Error once, check that we retry and not fail"""
+
+        responses = [{"json": {"error": {"code": 104}}}]
+
+        requests_mock.register_uri("GET", FacebookSession.GRAPH + f"/{FB_API_VERSION}/act_{account_id}/insights", responses)
+
+        with pytest.raises(FacebookRequestError):
+            stream = AdsInsights(api=api, start_date=datetime.now(), end_date=datetime.now())
+            list(stream.stream_slices(stream_state={}, sync_mode=None))

docs/integrations/sources/facebook-marketing.md

@@ -103,6 +103,7 @@ As a summary, custom insights allows to replicate only some fields, resulting in
 
 | Version | Date | Pull Request | Subject |
 | :--- | :--- | :--- | :--- |
+| 0.2.35  | 2022-02-18 | [10348](https://github.com/airbytehq/airbyte/pull/10348) | Add 104 error code to backoff triggers |
 | 0.2.34  | 2022-02-17 | [10180](https://github.com/airbytehq/airbyte/pull/9805) | Performance and reliability fixes |
 | 0.2.33  | 2021-12-28 | [10180](https://github.com/airbytehq/airbyte/pull/10180) | Add AdAccount and Images streams |
 | 0.2.32  | 2022-01-07 | [10138](https://github.com/airbytehq/airbyte/pull/10138) | Add `primary_key` for all insights streams. |