chore: resolve comments

Author: pnilan

airbyte_cdk/sources/declarative/manifest_declarative_source.py

@@ -143,10 +143,6 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
         if api_budget_model:
             self._constructor.set_api_budget(api_budget_model, config)
 
-        max_concurrent_async_jobs_model = self._source_config.get("max_concurrent_async_jobs")
-        if max_concurrent_async_jobs_model:
-            self._constructor.set_max_concurrent_async_jobs(max_concurrent_async_jobs_model, config)
-
         source_streams = [
             self._constructor.create_component(
                 DeclarativeStreamModel,

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -541,8 +541,10 @@ def __init__(
         )
         self._connector_state_manager = connector_state_manager or ConnectorStateManager()
         self._api_budget: Optional[Union[APIBudget, HttpAPIBudget]] = None
-        self._job_tracker: Optional[JobTracker] = self._create_async_job_tracker(
-            source_config=source_config
+        self._job_tracker: Optional[JobTracker] = (
+            self._create_async_job_tracker(source_config=source_config)
+            if source_config
+            else None
         )
 
     def _init_mappings(self) -> None:
@@ -2928,8 +2930,7 @@ def _get_download_retriever() -> SimpleRetrieverTestReadDecorator | SimpleRetrie
             download_target_extractor=download_target_extractor,
         )
 
-        if not self._job_tracker:
-            self._job_tracker = JobTracker(1)
+        self._job_tracker = JobTracker(1) if not self._job_tracker else self._job_tracker
 
         async_job_partition_router = AsyncJobPartitionRouter(
             job_orchestrator_factory=lambda stream_slices: AsyncJobOrchestrator(