refactor back to old way

Author: brianjlai

airbyte_cdk/sources/declarative/concurrent_declarative_source.py

@@ -36,7 +36,7 @@
 )
 from airbyte_cdk.sources.declarative.partition_routers import AsyncJobPartitionRouter
 from airbyte_cdk.sources.declarative.requesters import HttpRequester
-from airbyte_cdk.sources.declarative.retrievers import AsyncRetriever, SimpleRetriever
+from airbyte_cdk.sources.declarative.retrievers import AsyncRetriever, Retriever, SimpleRetriever
 from airbyte_cdk.sources.declarative.stream_slicers.declarative_partition_generator import (
     DeclarativePartitionFactory,
     StreamSlicerPartitionGenerator,
@@ -220,27 +220,15 @@ def _group_streams(
                 if self._is_datetime_incremental_without_partition_routing(
                     declarative_stream, incremental_sync_component_definition
                 ):
-                    retriever = declarative_stream.retriever
-
-                    # This is an optimization so that we don't invoke any cursor or state management flows within the
-                    # low-code framework because state management is handled through the ConcurrentCursor.
-                    if declarative_stream and isinstance(retriever, SimpleRetriever):
-                        # Also a temporary hack. In the legacy Stream implementation, as part of the read,
-                        # set_initial_state() is called to instantiate incoming state on the cursor. Although we no
-                        # longer rely on the legacy low-code cursor for concurrent checkpointing, low-code components
-                        # like StopConditionPaginationStrategyDecorator and ClientSideIncrementalRecordFilterDecorator
-                        # still rely on a DatetimeBasedCursor that is properly initialized with state.
-                        if retriever.cursor:
-                            stream_state = self._connector_state_manager.get_stream_state(
-                                stream_name=declarative_stream.name,
-                                namespace=declarative_stream.namespace,
-                            )
-                            retriever.cursor.set_initial_state(stream_state=stream_state)
-                        # We zero it out here, but since this is a cursor reference, the state is still properly
-                        # instantiated for the other components that reference it
-                        retriever.cursor = None
+                    stream_state = self._connector_state_manager.get_stream_state(
+                        stream_name=declarative_stream.name, namespace=declarative_stream.namespace
+                    )
+
+                    retriever = self._get_retriever(declarative_stream, stream_state)
 
-                    if isinstance(declarative_stream.retriever, AsyncRetriever):
+                    if isinstance(declarative_stream.retriever, AsyncRetriever) and isinstance(
+                        declarative_stream.retriever.stream_slicer, AsyncJobPartitionRouter
+                    ):
                         cursor = declarative_stream.retriever.stream_slicer.stream_slicer
 
                         if not isinstance(cursor, ConcurrentCursor):
@@ -357,7 +345,7 @@ def _group_streams(
                         )
                     )
 
-                    retriever = declarative_stream.retriever
+                    retriever = self._get_retriever(declarative_stream, stream_state)
 
                     partition_generator = StreamSlicerPartitionGenerator(
                         DeclarativePartitionFactory(
@@ -474,6 +462,28 @@ def _stream_supports_concurrent_partition_processing(
                             return False
         return True
 
+    @staticmethod
+    def _get_retriever(
+        declarative_stream: DeclarativeStream, stream_state: Mapping[str, Any]
+    ) -> Retriever:
+        retriever = declarative_stream.retriever
+
+        # This is an optimization so that we don't invoke any cursor or state management flows within the
+        # low-code framework because state management is handled through the ConcurrentCursor.
+        if declarative_stream and isinstance(retriever, SimpleRetriever):
+            # Also a temporary hack. In the legacy Stream implementation, as part of the read,
+            # set_initial_state() is called to instantiate incoming state on the cursor. Although we no
+            # longer rely on the legacy low-code cursor for concurrent checkpointing, low-code components
+            # like StopConditionPaginationStrategyDecorator and ClientSideIncrementalRecordFilterDecorator
+            # still rely on a DatetimeBasedCursor that is properly initialized with state.
+            if retriever.cursor:
+                retriever.cursor.set_initial_state(stream_state=stream_state)
+            # We zero it out here, but since this is a cursor reference, the state is still properly
+            # instantiated for the other components that reference it
+            retriever.cursor = None
+
+        return retriever
+
     @staticmethod
     def _select_streams(
         streams: List[AbstractStream], configured_catalog: ConfiguredAirbyteCatalog