call state migration in ConcurrentDeclarativeSource

Author: darynaishchenko

airbyte_cdk/sources/declarative/concurrent_declarative_source.py

@@ -224,6 +224,9 @@ def _group_streams(
                     stream_state = self._connector_state_manager.get_stream_state(
                         stream_name=declarative_stream.name, namespace=declarative_stream.namespace
                     )
+                    for state_migration in declarative_stream.state_migrations:
+                        if state_migration.should_migrate(stream_state):
+                            stream_state = state_migration.migrate(stream_state)
 
                     retriever = self._get_retriever(declarative_stream, stream_state)
 
@@ -331,6 +334,10 @@ def _group_streams(
                     stream_state = self._connector_state_manager.get_stream_state(
                         stream_name=declarative_stream.name, namespace=declarative_stream.namespace
                     )
+                    for state_migration in declarative_stream.state_migrations:
+                        if state_migration.should_migrate(stream_state):
+                            stream_state = state_migration.migrate(stream_state)
+
                     partition_router = declarative_stream.retriever.stream_slicer._partition_router
 
                     perpartition_cursor = (