chore(ci_visibility): fetch test data concurrently [recreated] (#12993)

Recreating #12972 because we
needed to revert PRs that were accidentally merged prematurely due to
required status checks getting deleted.

## Checklist
- [x] PR author has checked that all the criteria below are met
- The PR description includes an overview of the change
- The PR description articulates the motivation for the change
- The change includes tests OR the PR description describes a testing
strategy
- The PR description notes risks associated with the change, if any
- Newly-added code is easy to change
- The change follows the [library release note
guidelines](https://ddtrace.readthedocs.io/en/stable/releasenotes.html)
- The change includes or references documentation updates if necessary
- Backport labels are set (if
[applicable](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting))

## Reviewer Checklist
- [x] Reviewer has checked that all the criteria below are met 
- Title is accurate
- All changes are related to the pull request's stated goal
- Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes
- Testing strategy adequately addresses listed risks
- Newly-added code is easy to change
- Release note makes sense to a user of the library
- If necessary, author has acknowledged and discussed the performance
implications of this PR as reported in the benchmarks PR comment
- Backport labels are set in a manner that is consistent with the
[release branch maintenance
policy](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting)

Co-authored-by: Vítor De Araújo <[email protected]>

Author: erikayasuda

Diff for: ddtrace/internal/ci_visibility/recorder.py

@@ -1,3 +1,4 @@
+from concurrent.futures import ThreadPoolExecutor
 import json
 import os
 from pathlib import Path
@@ -607,31 +608,46 @@ def _start_service(self) -> None:
             tracer_filters += [TraceCiVisibilityFilter(self._tags, self._service)]  # type: ignore[arg-type]
             self.tracer.configure(trace_processors=tracer_filters)
 
-        if self.test_skipping_enabled():
-            self._fetch_tests_to_skip()
-            if self._itr_data is None:
-                log.warning("Failed to fetch skippable items, no tests will be skipped.")
-                return
-            log.info("Intelligent Test Runner skipping level: %s", "suite" if self._suite_skipping_mode else "test")
-            log.info("Skippable items fetched: %s", len(self._itr_data.skippable_items))
-            log.info("ITR correlation ID: %s", self._itr_data.correlation_id)
-
-        if CIVisibility.is_efd_enabled():
-            known_test_ids = self._fetch_known_tests()
-            if known_test_ids is None:
-                log.warning("Failed to fetch unique tests for Early Flake Detection")
+        def _task_fetch_tests_to_skip():
+            if self.test_skipping_enabled():
+                self._fetch_tests_to_skip()
+                if self._itr_data is None:
+                    log.warning("Failed to fetch skippable items, no tests will be skipped.")
+                    return
+                log.info("Intelligent Test Runner skipping level: %s", "suite" if self._suite_skipping_mode else "test")
+                log.info("Skippable items fetched: %s", len(self._itr_data.skippable_items))
+                log.info("ITR correlation ID: %s", self._itr_data.correlation_id)
+
+        def _task_fetch_known_tests():
+            if CIVisibility.is_efd_enabled():
+                known_test_ids = self._fetch_known_tests()
+                if known_test_ids is None:
+                    log.warning("Failed to fetch unique tests for Early Flake Detection")
+                else:
+                    self._known_test_ids = known_test_ids
+                    log.info("Unique tests fetched for Early Flake Detection: %s", len(self._known_test_ids))
             else:
-                self._known_test_ids = known_test_ids
-                log.info("Unique tests fetched for Early Flake Detection: %s", len(self._known_test_ids))
-        else:
-            if (
-                self._api_settings.early_flake_detection.enabled
-                and not ddconfig._test_visibility_early_flake_detection_enabled
-            ):
-                log.warning(
-                    "Early Flake Detection is enabled by API but disabled by "
-                    "DD_TEST_VISIBILITY_EARLY_FLAKE_DETECTION_ENABLED environment variable"
-                )
+                if (
+                    self._api_settings.early_flake_detection.enabled
+                    and not ddconfig._test_visibility_early_flake_detection_enabled
+                ):
+                    log.warning(
+                        "Early Flake Detection is enabled by API but disabled by "
+                        "DD_TEST_VISIBILITY_EARLY_FLAKE_DETECTION_ENABLED environment variable"
+                    )
+
+        def _task_fetch_test_management_tests():
+            if self._api_settings.test_management.enabled:
+                test_properties = self._fetch_test_management_tests()
+                if test_properties is None:
+                    log.warning("Failed to fetch quarantined tests from Test Management")
+                else:
+                    self._test_properties = test_properties
+
+        with ThreadPoolExecutor() as pool:
+            pool.submit(_task_fetch_tests_to_skip)
+            pool.submit(_task_fetch_known_tests)
+            pool.submit(_task_fetch_test_management_tests)
 
         if self._api_settings.flaky_test_retries_enabled and not asbool(
             os.environ.get("DD_CIVISIBILITY_FLAKY_RETRY_ENABLED", True)
@@ -641,13 +657,6 @@ def _start_service(self) -> None:
                 "DD_CIVISIBILITY_FLAKY_RETRY_ENABLED environment variable"
             )
 
-        if self._api_settings.test_management.enabled:
-            test_properties = self._fetch_test_management_tests()
-            if test_properties is None:
-                log.warning("Failed to fetch quarantined tests from Test Management")
-            else:
-                self._test_properties = test_properties
-
     def _stop_service(self) -> None:
         if self._should_upload_git_metadata and not self._git_client.metadata_upload_finished():
             log.debug("git metadata upload still in progress, waiting before shutting down")