feat(Low-Code Concurrent CDK): Make SimpleRetriever thread-safe so that different partitions can share the same SimpleRetriever #185
Conversation
… yielding to be simpler, fix tests, formatting, mypy errors
| for record in records_generator_fn(response): | ||
| last_page_size += 1 | ||
| last_record = record | ||
| yield record |
There was a problem hiding this comment.
This part is tricky. Before, used to manage _last_record and _last_page_size as fields on the SimpleRetriever and we updated it from within _parse_response().
However, to make this stateless, instead of trying to yield records and at the end returning a value which was messy, I moved all the assignment up to this level. In my testing, this seemed to be the equivalent functionality of setting last_response first then updating last_page_size and last_record as records are emitting. Just want to highlight this change a bit since it does subtly change the flow a bit
| @@ -204,6 +212,9 @@ class PaginatorTestReadDecorator(Paginator): | |||
| """ | |||
| In some cases, we want to limit the number of requests that are made to the backend source. This class allows for limiting the number of | |||
| pages that are queried throughout a read command. | |||
|
|
|||
| WARNING: This decorator is not currently thread-safe like the rest of the low-code framework because it has | |||
| an internal state to track the current number of pages counted so that it can exit early during a test read | |||
There was a problem hiding this comment.
Also highlighting this. I'd love for the connector builder server's usage of the PaginatorTestReadDecorator to be stateless too, but I couldn't think of a good way of making the page count limit cut avoid storing state. And I didn't want to block getting this fix out either. If anyone has a clever solution for this.
We probably need need to solve this when we move the builder server to use the concurrent source
| @@ -24,27 +24,32 @@ class Paginator(ABC, RequestOptionsProvider): | |||
| """ | |||
|
|
|||
| @abstractmethod | |||
| def reset(self, reset_value: Optional[Any] = None) -> None: | |||
There was a problem hiding this comment.
reseting can be deprecated now that these classes are stateless. We do however need to add the ability to get the initial token value for different types of pagination strategies
| @@ -518,7 +525,7 @@ def _parse_records( | |||
| stream_state: Mapping[str, Any], | |||
| records_schema: Mapping[str, Any], | |||
| stream_slice: Optional[StreamSlice], | |||
| ) -> Iterable[StreamData]: | |||
| ) -> Iterable[Record]: | |||
There was a problem hiding this comment.
this wasn't typed correctly because _parse_response returns an iterable of records. This is a more accurate type
brianjlai
changed the title
brianjlai
changed the title
📝 WalkthroughWalkthroughThe pull request introduces comprehensive modifications to pagination and retrieval mechanisms across multiple files in the Airbyte CDK. The changes primarily focus on enhancing the flexibility of pagination strategies, stream handling, and concurrency management. Key modifications include updating method signatures to support more dynamic token handling, removing reset methods, and refining how streams are grouped and processed in concurrent scenarios. Changes
Possibly related PRs
Suggested labels
Suggested reviewers
Would you like me to elaborate on any specific aspect of these changes? 🤔 The modifications seem quite comprehensive, wdyt? 🧐 Tip CodeRabbit's docstrings feature is now available as part of our Early Access Program! Simply use the command Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media? 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (48)
airbyte_cdk/sources/declarative/requesters/paginators/default_paginator.py (6)
167-167: Headers retrieval
This approach mirrors request params. WDYT about merging similar logic into a shared helper function?
176-176: Body data retrieval
Sticking to the same pattern is consistent. Would you like to store the body data in a local variable for debugging, wdyt?
187-199: Token injection
We inject the page token if it exists. Looks correct. Maybe log a warning iftokenis unexpectedly None, wdyt?
231-233: Reset page count
get_initial_token()re-initializes_page_count. Might it be clearer if_page_countwas renamed to_remaining_pagesor similar, wdyt?
250-251: Path delegation
We forward path calls as well. WDYT about caching the path for debug or logging?
Line range hint
287-291: Fallback retriever factory
We switch to_retriever_factory()if it's not async. Maybe we can handle concurrency-unsafe retrievers here more directly, wdyt?airbyte_cdk/sources/declarative/concurrent_declarative_source.py (4)
197-201: Retriever type check
We check if the retriever isSimpleRetrieverorAsyncRetriever. Perhaps unify both under a common interface to avoid conditionals, wdyt?
283-286: Factory method
This closure returns a sharedasync_retriever. Do we want to add doc comments clarifying concurrency usage, wdyt?
287-291: Fallback retriever factory
We delegate to_retriever_factory()if not async. Maybe unify the logic so a single method picks the right path automatically, wdyt?
296-296: Wrapping partition generator
This code sets up theDeclarativePartitionFactory. A docstring note about concurrency constraints might help, wdyt?airbyte_cdk/sources/declarative/retrievers/simple_retriever.py (9)
255-255: Record selection
We directly yield fromrecord_selector.select_records. This is nice and streamlined. Should we log the count of records, wdyt?
330-330: _read_pages method
You're usingrecords_generator_fnas a callback. Would you define a type alias for clarity, wdyt?
335-338: Initial token retrieval
We wrapinitial_tokeninto a dict if it's not None. Using a custom typed structure might be clearer, wdyt?
341-347: Tracking last record
We storelast_page_sizeandlast_recordto pass along. Maybe add a debug log here, wdyt?
369-369: _read_single_page
This approach focuses on a single page at a time. Would you unify it with_read_pagesto reduce duplication, wdyt?
390-390: Marking completion
We set{FULL_REFRESH_SYNC_COMPLETE_KEY: True}to mark the end of sync. Any interest in logging completion, wdyt?
392-400: Next page token fallback
If_next_page_tokenis None, we default to a "sync complete" token. Would short-circuiting iflast_page_sizeis 0 make sense, wdyt?
528-528: Accurate method signature
Updating_parse_recordsclarifies types. WDYT about adding a brief docstring?
572-572: Log formatter usage
We pass a custom log formatter for test reads, which is helpful. Would you also log the response status code, wdyt?airbyte_cdk/sources/declarative/requesters/paginators/strategies/pagination_strategy.py (1)
29-39: Extended next_page_token signature
We addedlast_page_token_valueto handle stateless pagination. Could we enhance the docstring to clarify how this differs from the token inresponse, wdyt?airbyte_cdk/sources/declarative/requesters/paginators/no_pagination.py (1)
65-70:next_page_tokenreturning an empty dictionary.
Currently, this returns an empty dictionary rather thanNone. Would returningNonefor a no-pagination scenario be more intuitive, or do you prefer an empty dictionary for consistency with other pagination strategies? wdyt?airbyte_cdk/sources/declarative/requesters/paginators/paginator.py (2)
34-38:next_page_tokensignature updated.
Addinglast_page_token_valueparameter gives more control over pagination status in concurrent settings. Is it worth explicitly documenting concurrency patterns here, or is that best kept in a separate concurrency guide? wdyt?
46-46: Expanded docstring to mentionlast_page_token_value.
Thanks for making this param clearer. Might we also add an example usage snippet, or is that unnecessary here? wdyt?airbyte_cdk/sources/declarative/requesters/paginators/strategies/stop_condition.py (1)
53-53: Comment about reverse order.
Your reasoning here is quite interesting. Would it help to clarify why many APIs return data in descending order, so other developers can better understand this assumption? wdyt?airbyte_cdk/sources/declarative/requesters/paginators/strategies/page_increment.py (2)
45-45: Returningstart_from_pageas theinitial_token.
This is a clear way to define the first page. Would you consider logging or validating thatstart_from_pageis a positive integer, just to catch configuration errors early? wdyt?
65-65: Validation for integer page tokens.
Throwing aValueErrorhere is good defensive coding. We might consider logging the contents oflast_page_token_valuefor easier debugging, wdyt?unit_tests/sources/declarative/requesters/paginators/test_offset_increment.py (1)
48-54: Consider documenting the new parameter in a docstring.
Defining the purpose and expected types oflast_page_token_valuecould help maintain clarity for future contributors. wdyt?unit_tests/sources/declarative/requesters/paginators/test_page_increment.py (1)
17-39: Expanded parameter list
Includinglast_page_token_valueallows the paginator to be stateless. This is a beneficial change for concurrency. Did we consider adding a scenario wherelast_page_token_valueis present but doesn't match the expected next page token? wdyt?airbyte_cdk/sources/declarative/requesters/paginators/strategies/cursor_pagination_strategy.py (2)
59-64: Clear documentation forinitial_token.
MarkingNoneas the default clarifies that the strategy starts based on external references (state or the first response). Maybe highlight in a class docstring that there is no explicit “cursor reset”? wdyt?
67-71: Updated signature withlast_page_token_value.
This keeps the interface consistent with the other pagination strategies. Any thoughts on adding a small log message iflast_page_token_valueis ignored in this strategy? wdyt?unit_tests/sources/declarative/test_concurrent_declarative_source.py (3)
654-655: Double-check concurrency count?
Lines 654–655 comment on the new async job stream, and we ensure the test for six concurrent streams. Would you like to also verify each stream type to confirm that all newly added streams are indeed part of this count? wdyt?
662-662: Clean enumeration of concurrent streams
You’re unpacking each stream to a unique variable, which helps with clarity. Perhaps confirm in a follow-up test that each element is indeed configured as expected (e.g., correct path, paginator, etc.)? wdyt?
678-678: Ensure synchronous streams are declared
You assert there is exactly one synchronous stream. Would you like to add a short docstring indicating why this particular set of streams runs synchronously, for future clarity? wdyt?unit_tests/sources/declarative/test_manifest_declarative_source.py (2)
1368-1368: Call signature consistency
The call signature for_fetch_next_pageincludes{},{}, andNone. Would you prefer stating parameters with explicit naming for clarity in future test maintenance? wdyt?
1538-1545: Refining multi-page calls
Here, we see multiple calls to_fetch_next_page({…})withnext_page_tokenparameters across lines 1538–1545. Would you be open to verifying that each new call is passing along the correct updated token from the prior response, especially if multiple tokens are chained? wdyt?airbyte_cdk/sources/declarative/requesters/paginators/strategies/offset_increment.py (2)
70-74: Great job passing inlast_page_token_value.
This parameterization is a clean, stateless approach for handling page tokens. Do you think a more descriptive name (e.g.,previous_offset) would enhance clarity, wdyt?
84-94: Validating the page token is an excellent safeguard.
Raising aValueErrorfor non-integer tokens will prevent subtle pagination bugs. Have you considered using a custom exception class to allow further debugging or logging, wdyt?unit_tests/sources/declarative/requesters/paginators/test_default_paginator.py (1)
211-216: Explicitly passingNonefornext_page_tokenis clear.
This makes it obvious we're starting from the initial page. Would you consider adding a docstring clarifying that passingNoneindicates a “fresh start,” wdyt?unit_tests/sources/declarative/retrievers/test_simple_retriever.py (10)
6-7: Nice usage ofpartialand type hints.
They improve code readability. Would you like to also consider a type alias forMapping[str, Any]for further clarity, wdyt?
29-32: Imports forCursorPaginationStrategyandPageIncrementlook good.
They align with the new stateless approach. Any thoughts on grouping these pagination strategies under a single test suite for maintainability, wdyt?
40-40:TypeTransformerusage
It's nice to see better transformations for result data. Would you consider adding tests specifically for transformation logic, wdyt?
62-62: Defaulting initial token toNoneis consistent.
This ensures that calling code clearly manages the start state. Have you considered a sentinel value instead ofNonefor additional clarity, wdyt?
66-66: Returning empty headers by default
It’s straightforward. Might it be useful to log if there are no headers, or do you prefer silent defaults, wdyt?
118-121: Testing_next_page_tokenwith these parameters
This ensures coverage for concurrency scenarios. Would you add a scenario for multi-page transitions, wdyt?
394-394:paginator.get_initial_tokenis mocked out
Skipping calls indicates you’re ensuring no resets happen if the stream is complete. Maybe a docstring clarifies this logic, wdyt?
668-668: Implementing_paginator_pathoverride
Nice approach to unify path logic from both the requester and paginator. Have you considered a fallback or default path scenario, wdyt?
848-850: New tests for logging
You confirm that your custom log formatter is invoked. A quick question: do you see a need for different logging behaviors based on environment (dev vs. prod), wdyt?
942-1057:test_retriever_is_stateless
Verifying no internal state is definitely crucial for concurrency. Great test approach! Could you expand coverage to confirm correct behavior across different page size inputs, wdyt?
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (19)
airbyte_cdk/sources/declarative/concurrent_declarative_source.py(3 hunks)airbyte_cdk/sources/declarative/requesters/paginators/default_paginator.py(7 hunks)airbyte_cdk/sources/declarative/requesters/paginators/no_pagination.py(2 hunks)airbyte_cdk/sources/declarative/requesters/paginators/paginator.py(1 hunks)airbyte_cdk/sources/declarative/requesters/paginators/strategies/cursor_pagination_strategy.py(1 hunks)airbyte_cdk/sources/declarative/requesters/paginators/strategies/offset_increment.py(2 hunks)airbyte_cdk/sources/declarative/requesters/paginators/strategies/page_increment.py(1 hunks)airbyte_cdk/sources/declarative/requesters/paginators/strategies/pagination_strategy.py(2 hunks)airbyte_cdk/sources/declarative/requesters/paginators/strategies/stop_condition.py(1 hunks)airbyte_cdk/sources/declarative/retrievers/simple_retriever.py(10 hunks)unit_tests/sources/declarative/requesters/paginators/test_cursor_pagination_strategy.py(2 hunks)unit_tests/sources/declarative/requesters/paginators/test_default_paginator.py(3 hunks)unit_tests/sources/declarative/requesters/paginators/test_no_paginator.py(1 hunks)unit_tests/sources/declarative/requesters/paginators/test_offset_increment.py(1 hunks)unit_tests/sources/declarative/requesters/paginators/test_page_increment.py(1 hunks)unit_tests/sources/declarative/requesters/paginators/test_stop_condition.py(2 hunks)unit_tests/sources/declarative/retrievers/test_simple_retriever.py(10 hunks)unit_tests/sources/declarative/test_concurrent_declarative_source.py(3 hunks)unit_tests/sources/declarative/test_manifest_declarative_source.py(4 hunks)
🔇 Additional comments (51)
airbyte_cdk/sources/declarative/requesters/paginators/default_paginator.py (9)
115-123: Use caution with the new get_initial_token approach
We've introduced get_initial_token() returning self.pagination_strategy.initial_token. This is consistent with the new stateless approach. Would you consider clarifying the docstring about checkpointing, wdyt?
126-139: Focus on next_page_token usage
The parameter last_page_token_value is a nice addition to ensure continuity between requests. Maybe we could handle the scenario if there's a mismatch between last_page_token_value and response, wdyt?
143-147: Path rewriting logic
Replacing the base URL with an empty string is clever for path extraction. If the URL doesn't contain the base substring, do we handle it gracefully, wdyt?
158-158: Good use of _get_request_options
This usage unifies request parameter handling. WDYT about logging if we skip certain params?
185-185: JSON body retrieval
Following the same approach as the other request options. Looks great!
236-240: Decorator next_page_token
We skip pagination after reaching the page limit. Do you want to log how many pages were actually read, wdyt?
246-248: Delegation
We delegate the next_page_token to the decorated paginator. That seems convenient.
Line range hint 277-286: AsyncRetriever approach
We share the async_retriever across partitions. Should we add a test for concurrency safety, wdyt?
Line range hint 296-296: Partition generator
We pass the retriever factory through. Everything looks consistent.
airbyte_cdk/sources/declarative/concurrent_declarative_source.py (1)
221-225: Async job stream detection
We detect AsyncRetriever by type. Would you consider validating that the config indeed supports async flows, wdyt?
airbyte_cdk/sources/declarative/retrievers/simple_retriever.py (8)
9-9: Import statement update
We extend typing usage to include more types. Maybe consider forward refs if needed, wdyt?
236-242: Paginator path usage
We pass the next_page_token to _paginator.path(), ensuring the path can reflect the correct token. Maybe add a fallback if _paginator.path() is None, wdyt?
273-279: Enhanced next_page_token parameters
Including last_page_size, last_record, and last_page_token_value makes pagination more robust. Would you handle partial pages differently, wdyt?
287-292: Delegate to the pagination strategy
We pass all needed params to pagination_strategy.next_page_token. WDYT about logging the token for debug?
352-360: Deriving next page token
We pass the old token as last_page_token_value. The approach looks consistent.
373-379: Initial token from state
We try stream_state.get("next_page_token") first, or call get_initial_token(). What if the stored token is invalid, wdyt?
380-387: Single-page record loop
The logic to track last_record remains symmetrical with multi-page usage. Nicely done!
435-437: Full refresh early exit
If the prior job was successful, we skip new fetching. This is an efficient optimization.
unit_tests/sources/declarative/requesters/paginators/test_no_paginator.py (1)
12-12: Additional argument
We pass None as the fourth parameter to next_page_token. That aligns with the updated signature. All good, wdyt?
airbyte_cdk/sources/declarative/requesters/paginators/strategies/pagination_strategy.py (1)
7-7: Typing improvements
We import Mapping for better type clarity. Would you consider using collections.abc.Mapping for Python 3.9+, wdyt?
airbyte_cdk/sources/declarative/requesters/paginators/no_pagination.py (2)
22-22: Method signature updated to accept a page token.
It looks like this method now aligns with the new interface that requires a next_page_token. Since this class never paginates, returning None seems perfectly logical. Do you agree, or do you think returning an empty string might be clearer for some APIs? wdyt?
61-62: Initial token returning None.
This clearly communicates that there is no established starting token. Seems consistent with the “no pagination” concept. Any concerns about potential confusion for newcomers? wdyt?
airbyte_cdk/sources/declarative/requesters/paginators/paginator.py (2)
27-29: New get_initial_token method.
This method clarifies how to retrieve a starting token for paginators. It’s a welcome addition that keeps classes stateless. Do you think we need more documentation examples for typical usage? wdyt?
52-52: path now uses next_page_token.
Providing the next_page_token to the path method is a strong step toward full statelessness. Do you see any value in also accepting other optional context values here, or is that beyond scope? wdyt?
airbyte_cdk/sources/declarative/requesters/paginators/strategies/stop_condition.py (2)
47-51: next_page_token signature supports last_page_token_value.
This addition looks consistent with the broader changes. Might it be beneficial to log or validate last_page_token_value inside this decorator if it doesn’t meet certain conditions? wdyt?
57-59: Delegate’s next_page_token call.
Forwarding the parameters to _delegate is straightforward. Any thoughts on adding specialized post-processing before returning? wdyt?
airbyte_cdk/sources/declarative/requesters/paginators/strategies/page_increment.py (3)
49-53: next_page_token signature now includes last_page_token_value.
This aligns well with the concurrency-friendly approach. Are you considering adding debug logs if last_page_token_value is unexpectedly None? wdyt?
58-63: Handling None page token.
Great approach assuming the first request was already made. However, do you foresee any case where None might appear mid-stream in edge scenarios (e.g., errors)? Would a fallback or error help with diagnostics? wdyt?
68-68: Incrementing the page token.
Simply returning last_page_token_value + 1 is very straightforward. Is there any edge case where the page token could approach a maximum integer limit, or do you assume that’s improbable for typical usage? wdyt?
unit_tests/sources/declarative/requesters/paginators/test_offset_increment.py (3)
17-44: Nice parameterization for last_page_token_value.
It looks consistent with the new signature in the OffsetIncrement class. Adding this parameter keeps the tests aligned with the new stateless approach. Just to be sure, do we need an additional test to confirm that a None value for last_page_token_value behaves correctly? wdyt?
64-66: Good use of the new signature
The call to next_page_token with last_page_token_value shows that the code is consistent with the recently introduced parameter.
69-73: Verifying statelessness
The repeated invocation of next_page_token helps confirm that no internal state depends on a prior call’s side effects. This is a valuable test. Great work!
unit_tests/sources/declarative/requesters/paginators/test_stop_condition.py (2)
89-91: Ensuring concurrency safety
The updated call to next_page_token with last_page_token_value is in line with the new pattern. Do we want to add a separate test explicitly covering concurrency scenarios? wdyt?
[approve]
105-106: Consistent parameter usage
The additional None parameter here aligns with the refactored interface. Maintaining consistency across these test calls strengthens the overall verification. Nicely done!
unit_tests/sources/declarative/requesters/paginators/test_page_increment.py (4)
44-50: Straightforward function signature
Declaring these parameters explicitly clarifies the test logic. This approach of passing everything in makes each test scenario very transparent.
55-55: Verifying start_from_page
Asserting against paginator_strategy.start_from_page instead of a private attribute is more robust and future-proof. Good practice!
63-65: Checking next_page_token
Using a real HTTP response ensures we’re testing the logic with actual request/response flows. This is a strong setup.
68-72: Confirming repeated calls remain consistent
It’s great that we verify calling next_page_token multiple times yields the same token, preventing internal state issues.
unit_tests/sources/declarative/test_concurrent_declarative_source.py (2)
674-675: Validate concurrency configuration
Asserting the type and name ensures the stream is correctly identified as async. Would you consider adding a check that ensures the async_job_stream is truly non-blocking or concurrency-ready, if relevant? wdyt?
1461-1464: Confirm concurrency vs synchronous grouping
These comments and assertions verify that certain streams are concurrent vs. synchronous. Would you like to add a test scenario ensuring the boundary condition (e.g., if a stream gets incremental + partitioning, it moves to synchronous) is also tested? wdyt?
unit_tests/sources/declarative/test_manifest_declarative_source.py (2)
1281-1281: Confirm _fetch_next_page calls
You’re calling _fetch_next_page({}, {}, None) once. Do you anticipate a scenario where extra parameters (like next_page_token) might appear even for no-pagination tests? wdyt?
1452-1452: Flatten fields test
The test checks a single call to _fetch_next_page({}, {}, None). Is that sufficient to verify all flattened scenarios, or do we need more coverage for multi-page results? wdyt?
airbyte_cdk/sources/declarative/requesters/paginators/strategies/offset_increment.py (1)
66-66: Removing reliance on _offset is a nice simplification.
By returning 0 for the initial token, you ensure pagination starts at zero when inject_on_first_request is True. This is straightforward and improves clarity, wdyt?
unit_tests/sources/declarative/requesters/paginators/test_cursor_pagination_strategy.py (1)
15-15: Switching last_record to a Record instance is a solid improvement.
It helps ensure consistent typing and usage. Do you plan to adopt this Record pattern in other pagination tests for consistency, wdyt?
Also applies to: 83-83
unit_tests/sources/declarative/requesters/paginators/test_default_paginator.py (2)
284-285: Using paginator.next_page_token(...) with None ensures correct interpolation.
This is consistent with your stateless design. Do you foresee any edge cases where None might conflict with a default token, wdyt?
335-338: Retrieving the initial token from paginator is well-structured.
It clearly separates initialization logic from normal pagination. Perhaps re-check whether you need more test coverage for different initial tokens, wdyt?
unit_tests/sources/declarative/retrievers/test_simple_retriever.py (5)
16-17: Using JsonDecoder consistently is a good approach.
It keeps the decoding strategy uniform across tests. Do you plan to parameterize the decoder if you introduce other formats, wdyt?
78-81: Storing last_page_size, last_record, and last_page_token_value
Explicitly passing these around is great for concurrency. Does the test address edge cases like partial pages or empty records, wdyt?
419-419: get_initial_token.assert_not_called()
This is a valid assertion to confirm that no additional pagination logic was triggered. If you expand concurrency scenarios, remember to re-check these call counts, wdyt?
851-890: test_retriever_last_page_size_for_page_increment
This is a good addition for verifying multi-page data retrieval. Do you also want to test if an unexpected smaller page size triggers a stop condition, wdyt?
892-940: test_retriever_last_record_for_page_increment
You’re validating that the new CursorPaginationStrategy handles last_record properly. Would it be worthwhile to test merging logic if multiple records claim to be the “last” record, wdyt?
[approve]
airbyte_cdk/sources/declarative/requesters/paginators/default_paginator.py
Show resolved
Hide resolved
brianjlai
changed the title
What
Right now, because the SimpleRetriever has an internally managed and modified state, we cannot have multiple partitions running at the same time share the same SimpleRetriever. Otherwise we might run into data loss because the state of the SimpleRetriever (things like page number, etc) are modified by different partitions. We've seen this problem arise in some connections (link issue) and we have temporarily solved this by having every Partition instantiate it's own
SimpleRetriever.This however is very inefficient for a couple reasons like needing to perform auth for every partition (link issue). And this is a hard blocker on AsyncRetriever which must be shared across partitions in order to manage the shared job repository.
This PR replaces an internal state for
SimpleRetrieverand all of its dependencies which are manged via atokenfield and makes all methods stateless by relying on passing parameterized values for thenext_page_tokeninstead.How
This PR basically makes the following changes for thread safety: to the underlying implementations of the
DefaultPaginatorand all the pagination strategies so that they no longer have internal fields and instead pass required information via parameters to be statelessSimpleRetrieverclass had internal fields that were removed and replaced by local method variables likelast_record, andlast_page_size. These are passed to and from methods local to the initialread_records()invocationnext_page_tokenbecause we need the prior value to get the new value in a stateless manner. This will impact custom components that will no longer adhere to the original interface. More on that below.PaginationStrategyandDefaultPaginatorcomponents is removed because there is no longer an internal state fieldI also added in the changes to allow
AsyncRetrieverto instantiated within the concurrent framework because I want to test some of this functionality with sendgrid which needs these changes.Other notes
PaginatorandPaginationStrategymethod interfaces to now include additional parameters and our internal code now passes extra parameterssource-freshdesk: custom pagination strategysource-linkedin-ads: custom retrieversource-mixpanel: custom pagination strategysource-monday: custom pagination strategysource-zendesk-chat: custom pagination strategysource-the-guardian-api: custom pagination strategyBecause there really seems to only be one very low usage community connector that would be affected by the breaking change, I am proposing we make this a minor change so that all of our connectors don't get pinned behind a breaking change and can continue to receive our future changes that will improve adoption of concurrency. I'm happy to pick up this discussion in the PR if there's opposition
Summary by CodeRabbit
New Features
Bug Fixes
Tests