Update unit test for StateDelegatingRetriever

Author: lazebnyi

airbyte_cdk/sources/declarative/concurrent_declarative_source.py

@@ -162,6 +162,7 @@ def read(
         else:
             filtered_catalog = catalog
 
+        # It is no need run read for synchronous streams if they are not exists.
         if not filtered_catalog.streams:
             return
 

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -3152,7 +3152,7 @@ definitions:
         type: object
         additionalProperties: true
   StateDelegatingRetriever:
-    description: Test state condition retriever.
+    description: (This component is experimental. Use at your own risk.) Orchestrate the retriever's usage based on the state value.
     type: object
     required:
       - type
@@ -3168,14 +3168,14 @@ definitions:
         default: false
       full_refresh_retriever:
         title: Retriever
-        description: Component used to coordinate how records are extracted across stream slices and request pages.
+        description: Component used to coordinate how records are extracted across stream slices and request pages when the state is empty or not provided.
         anyOf:
           - "$ref": "#/definitions/AsyncRetriever"
           - "$ref": "#/definitions/CustomRetriever"
           - "$ref": "#/definitions/SimpleRetriever"
       incremental_retriever:
         title: Retriever
-        description: Component used to coordinate how records are extracted across stream slices and request pages.
+        description: Component used to coordinate how records are extracted across stream slices and request pages when the state provided.
         anyOf:
           - "$ref": "#/definitions/AsyncRetriever"
           - "$ref": "#/definitions/CustomRetriever"

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -2251,14 +2251,14 @@ class StateDelegatingRetriever(BaseModel):
     full_refresh_retriever: Union[AsyncRetriever, CustomRetriever, SimpleRetriever] = (
         Field(
             ...,
-            description="Component used to coordinate how records are extracted across stream slices and request pages.",
+            description="Component used to coordinate how records are extracted across stream slices and request pages when the state is empty or not provided.",
             title="Retriever",
         )
     )
     incremental_retriever: Union[AsyncRetriever, CustomRetriever, SimpleRetriever] = (
         Field(
             ...,
-            description="Component used to coordinate how records are extracted across stream slices and request pages.",
+            description="Component used to coordinate how records are extracted across stream slices and request pages when the state provided.",
             title="Retriever",
         )
     )

airbyte_cdk/sources/streams/concurrent/partition_enqueuer.py

@@ -55,7 +55,6 @@ def generate_partitions(self, stream: AbstractStream) -> None:
                 #
                 # Also note that prune_to_validate_has_reached_futures_limit has a lock while pruning which might create a bottleneck in
                 # terms of performance.
-
                 while self._thread_pool_manager.prune_to_validate_has_reached_futures_limit():
                     time.sleep(self._sleep_time_in_seconds)
                 self._queue.put(partition)