diff --git a/sdk/python/tests/utils/cli_repo_creator.py b/sdk/python/tests/utils/cli_repo_creator.py index 03c0976e880..48491610174 100644 --- a/sdk/python/tests/utils/cli_repo_creator.py +++ b/sdk/python/tests/utils/cli_repo_creator.py @@ -63,26 +63,43 @@ def _subprocess_env(self) -> dict[str, str]: env["PYTHONPATH"] = os.pathsep.join(parent_paths) return env - def run(self, args: List[str], cwd: Path) -> subprocess.CompletedProcess: + def run( + self, args: List[str], cwd: Path, attempts: int = 2 + ) -> subprocess.CompletedProcess: # Apply a conservative timeout to prevent CI hangs from Dask atexit-handler # stalls or other subprocess blockages. timeout = 120 - try: - return subprocess.run( - [sys.executable, cli.__file__] + args, - cwd=cwd, - capture_output=True, - timeout=timeout, - env=self._subprocess_env(), - ) - except subprocess.TimeoutExpired: - return subprocess.CompletedProcess( - args=[sys.executable, cli.__file__] + args, - returncode=-1, - stdout=b"", - stderr=f"Command timed out after {timeout}s: {args}".encode(), - ) + # Commands routed through here (e.g. `apply`) do not materialize, so they + # cannot trigger the Dask atexit hang described in the module docstring. A + # timeout is therefore almost always transient slowness — a cold child + # interpreter importing Feast (pandas/pyarrow/...) on a loaded CI runner, + # which has been observed to exceed the timeout on the slower macOS + # runners. Retry once so a single slow cold start does not flake the test: + # the retry runs with warm OS/page caches and typically finishes quickly, + # while a genuine repeated stall still fails once attempts are exhausted. + full_args = [sys.executable, cli.__file__] + args + result = None + for attempt in range(1, attempts + 1): + try: + return subprocess.run( + full_args, + cwd=cwd, + capture_output=True, + timeout=timeout, + env=self._subprocess_env(), + ) + except subprocess.TimeoutExpired: + result = subprocess.CompletedProcess( + args=full_args, + returncode=-1, + stdout=b"", + stderr=( + f"Command timed out after {timeout}s " + f"(attempt {attempt}/{attempts}): {args}" + ).encode(), + ) + return result def run_with_output(self, args: List[str], cwd: Path) -> Tuple[int, bytes]: is_teardown = "teardown" in args