[ISSUE #6829] update salesforce to support partitioned state #36942
Conversation
|
The latest updates on your projects. Learn more about Vercel for Git ↗︎ 1 Ignored Deployment
|
octavia-squidington-iii
added
area/connectors
octavia-squidington-iii
added
the
area/documentation
| @@ -404,131 +406,6 @@ def configure_request_params_mock(stream_1, stream_2): | |||
| stream_2.request_params.return_value = {"q": "query"} | |||
|
|
|||
|
|
|||
| def test_rate_limit_bulk(stream_config, stream_api, bulk_catalog, state): | |||
There was a problem hiding this comment.
They are very hard to maintain as they rely on private things to run for example page_size. Also, the way they are instantiated makes then not have the ConcurrentCursor as it relies on aggressively re-implementing source methods (like source.streams = Mock()). We will move these to mock server tests. We have already added a test than ensure it retries on 406 so we know the integration with the error handling works so I'm not too worried in removing this test specifically.
| @@ -949,15 +826,15 @@ def test_bulk_stream_error_on_wait_for_job(requests_mock, stream_config, stream_ | |||
| @freezegun.freeze_time("2023-01-01") | |||
| @pytest.mark.parametrize( | |||
| "lookback, stream_slice_step, expected_len_stream_slices, expect_error", | |||
| [(None, "P30D", 0, True), (0, "P30D", 158, False), (10, "P1D", 4732, False), (10, "PT12H", 9463, False), (-1, "P30D", 0, True)], | |||
There was a problem hiding this comment.
Removing the tests as they are kind of weird cases where either lookback is invalid or "we change lookback even though it is a constant". We have moved some of those tests to test_slice_generation.py
|
I'll add one test tomorrow morning to show that the new code supports the two state formats |
airbyte-integrations/connectors/source-salesforce/metadata.yaml
Outdated
Show resolved
Hide resolved
airbyte-integrations/connectors/source-salesforce/source_salesforce/source.py
Outdated
Show resolved
Hide resolved
maxi297
deleted the
issue-6829/update-source-salesforce-to-have-partitioned-state
branch
What
Addressing https://github.com/airbytehq/airbyte-internal-issues/issues/6829 to avoid stuck syncs.
Basically, syncs would get stuck when there was an error in a thread because we would try to shutdown the threads and threads that were already active would continue to sync without anything to consume the queue. To avoid that, we will run all the threads without breaking the main thread and save the state as a partitioned one so we only have to retry the slices that failed.
How
__init__have breaking changes)stream_slicesimplementationManual Testing
Compared output for:
🚨 User Impact 🚨
This is a breaking change as the state format will change. It does not need opt-in mechanism because the new version of the connector can understand the previous state format. However we can't revert this change as the old version doesn't understand the state of the new version.
This should allow threads to continue sync records even though one failed.