Add optional object-level retention support to GcsToGcsOperator

[chrisqiqiu]

closes: #60853

Reopens #64755 (the original PR was auto-closed because I accidentally deleted my fork — restored the branch and resubmitting with the same commits).

Add optional retain_until_time and retention_mode parameters to
GCSHook.rewrite() and GcsToGcsOperator, allowing object-level
retention to be applied to destination objects immediately after
copy/move operations.

This eliminates the need for a two-step "copy → set retention" workflow,
which is error-prone in compliance-driven pipelines where partial-success
can leave objects unprotected.

Changes:

• GCSHook.rewrite() — accepts retain_until_time and retention_mode,
  applies retention via blob.patch() after the rewrite completes
• GCSToGCSOperator — accepts the same parameters, passes them through
  _copy_single_object() to the hook. Retention kwargs are only passed
  when retain_until_time is set (fully backward compatible)
• Unit tests for both hook and operator covering retention, default mode,
  and no-retention cases

Testing:

• All existing tests passed for providers/google/tests/unit/google/cloud/hooks/test_gcs.py and providers/google/tests/unit/google/cloud/transfers/test_gcs_to_gcs.py

• Validated against real GCS with a retention-enabled bucket — retention
  mode and retain_until_time correctly applied and verified

---

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

• Yes (please specify the tool below)

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[SameerMesiah97]

Good PR. I only have 2 minor comments.

[shahar1]

LGTM, please address Sameer's comments

[SameerMesiah97]

LGTM, please address Sameer's comments

It looks good to me now but can you trigger CI?

[chrisqiqiu]

Thanks! I don't have permission to trigger CI. Could a maintainer approve the workflow run?

[SameerMesiah97]

Thanks! I don't have permission to trigger CI. Could a maintainer approve the workflow run?

I was asking @shahar1.

[shahar1]

Thanks! I don't have permission to trigger CI. Could a maintainer approve the workflow run?

I was asking @shahar1.

Done :)

[SameerMesiah97]

I see a failing test but it looks unrelated. Can CI be triggered again to see it was a flake?

[chrisqiqiu]

The failing test is TestBaseDatabricksHook::test_a_get_federated_token_with_projected_volume. It's a flaky timestamp comparison off by one second (1752325201 vs 1752325200) in the Databricks provider. looks like a race between when the token expiry is computed vs. when it's asserted. It's unrelated to my change. I only touched the Google provider's GCS code. The Google provider tests I changed in this PR shows OK for Test: Providers[google] https://github.com/apache/airflow/actions/runs/25255184172/job/74055520507?pr=66204#step:12:4669 . @shahar1
Could you pls re-run the failed job ?

[boring-cyborg]

Awesome work, congrats on your first merged pull request! You are invited to check our Issue Tracker for additional contributions.