State can be missing

Author: marianob-span1

airbyte_cdk/sources/declarative/concurrent_declarative_source.py

@@ -135,7 +135,8 @@ def read(
         catalog: ConfiguredAirbyteCatalog,
         state: Optional[List[AirbyteStateMessage]] = None,
     ) -> Iterator[AirbyteMessage]:
-        self._connector_state_manager = ConnectorStateManager(state=state)
+        if state:
+            self._connector_state_manager = ConnectorStateManager(state=state)
         concurrent_streams, _ = self._group_streams(config=config)
 
         # ConcurrentReadProcessor pops streams that are finished being read so before syncing, the names of