move stream_state calculation before calling create_concurrent_cursor_from_perpartition_cursor

Author: aldogonzalez8

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -1485,10 +1485,6 @@ def create_concurrent_cursor_from_perpartition_cursor(
             )
         )
 
-        if not stream_state:
-            stream_state = self._connector_state_manager.get_stream_state(
-                stream_name, stream_namespace
-            )
 
         stream_state = self.apply_stream_state_migrations(stream_state_migrations, stream_state)
         # Per-partition state doesn't make sense for GroupingPartitionRouter, so force the global state
@@ -1999,14 +1995,19 @@ def _build_incremental_cursor(
     ) -> Optional[StreamSlicer]:
         if model.incremental_sync and stream_slicer:
             if model.retriever.type == "AsyncRetriever":
+                stream_name = model.name or ""
+                stream_namespace = None
+                stream_state = self._connector_state_manager.get_stream_state(
+                    stream_name, stream_namespace
+                )
                 return self.create_concurrent_cursor_from_perpartition_cursor(  # type: ignore # This is a known issue that we are creating and returning a ConcurrentCursor which does not technically implement the (low-code) StreamSlicer. However, (low-code) StreamSlicer and ConcurrentCursor both implement StreamSlicer.stream_slices() which is the primary method needed for checkpointing
                     state_manager=self._connector_state_manager,
                     model_type=DatetimeBasedCursorModel,
                     component_definition=model.incremental_sync.__dict__,
-                    stream_name=model.name or "",
-                    stream_namespace=None,
+                    stream_name=stream_name,
+                    stream_namespace=stream_namespace,
                     config=config or {},
-                    stream_state={},
+                    stream_state=stream_state,
                     partition_router=stream_slicer,
                 )