Honor download_num_attempts and upload_num_attempts in GCSTimeSpanFileTransformOperator - #72953

GCSTimeSpanFileTransformOperator accepts download_num_attempts and upload_num_attempts ("Number of attempts to try to download/upload a single file"), but since #62196 moved the transfers from GCSHook.download() / GCSHook.upload() to direct blob.download_to_filename() / blob.upload_from_filename() calls, both values are stored and never read. Every blob gets exactly one attempt regardless of the setting, so a transient GoogleCloudError either fails the task or, with *_continue_on_fail, silently drops that file.

This restores the retry behaviour the operator had through the hook: a small _run_with_attempts helper retries GoogleCloudError up to the configured number of attempts with the hook's exponential back-off (2**attempt seconds), and both worker functions go through it. The default of one attempt keeps today's behaviour, and the *_continue_on_fail handling around the futures is unchanged.

Changes

• providers/google/src/airflow/providers/google/cloud/operators/gcs.py: _run_with_attempts helper; _download and _upload honor download_num_attempts and upload_num_attempts
• providers/google/tests/unit/google/cloud/operators/test_gcs.py: test_download_honors_download_num_attempts and test_upload_honors_upload_num_attempts (retry, give-up and single-attempt cases; time.sleep is patched and its calls asserted), plus a shared _setup_transform_process helper

Testing

• providers/google: tests/unit/google/cloud/operators/test_gcs.py (TestGCSTimeSpanFileTransformOperator, 28 tests)
• mypy on the changed module, prek hooks on the changed files

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Code (Claude Fable 5.1) following the guidelines. I reviewed and understand all changes; the tests were run locally as listed above.

🤖 Generated with Claude Code