bingqin2 commented on issue #74459:
URL: https://github.com/apache/airflow/issues/74459#issuecomment-6084344914
Tested #74305 and #72953 in `apache-airflow-providers-google==22.7.0rc1`.
Both work as expected.
Setup: a clean virtualenv with Python 3.12.13, `apache-airflow==3.3.2` and
the RC from PyPI. I ran the same checks against 22.6.0 for comparison. I don't
have a GCP project, so the BigQuery job states and the GCS blob calls are
faked. The triggers, the operator, the tenacity retry policy and the transform
subprocess are the installed code. shahar1's before/after run against real
BigQuery is in #74356.
| Check | 22.6.0 | 22.7.0rc1 |
|---|---|---|
| #74305: value check, job `RUNNING` on the first poll | fails with `Job
running` | keeps polling, succeeds |
| #74305: interval check, first job `RUNNING` on the first poll | fails with
`Job completed` (the second job's message) | keeps polling, succeeds |
| #74305: interval check, first job fails while the second runs | reports
`Job running` | reports the first job's error |
| #72953: download fails twice, `download_num_attempts=3` | fails after 1
call | succeeds on call 3, after 2 s and 4 s waits |
| #72953: upload fails once, `upload_num_attempts=2` | fails after 1 call |
succeeds on call 2 |
| #72953: download fails 3 times, `download_num_attempts=3` | raises after 1
call | raises after 3 calls |
<details><summary>Script</summary>
```python
"""Check #74305 and #72953 against an installed
apache-airflow-providers-google wheel.
No GCP access: BigQuery job states and GCS blob operations are faked,
everything else
(triggers, operator, tenacity retry policy, transform subprocess) is the
installed code.
"""
from __future__ import annotations
import asyncio
import importlib.metadata
import logging
import os
import stat
import sys
import tempfile
from pathlib import Path
from unittest import mock
import pendulum
from google.api_core.exceptions import ServiceUnavailable
logging.basicConfig(level=logging.WARNING, format=" log: %(message)s")
VERSION = importlib.metadata.version("apache-airflow-providers-google")
print(f"apache-airflow {importlib.metadata.version('apache-airflow')}, "
f"apache-airflow-providers-google {VERSION}, Python
{sys.version.split()[0]}")
from airflow.providers.google.cloud.triggers.bigquery import ( # noqa: E402
BigQueryIntervalCheckTrigger,
BigQueryValueCheckTrigger,
)
RUNNING = {"status": "running", "message": "Job running"}
DONE = {"status": "success", "message": "Job completed"}
FAILED = {"status": "error", "message": "Syntax error: Unexpected keyword
FROM at [1:8]"}
def run_trigger(trigger, job_states):
with
mock.patch("airflow.providers.google.cloud.triggers.bigquery.BigQueryAsyncHook")
as hook_cls:
hook = hook_cls.return_value
hook.get_job_status = mock.AsyncMock(side_effect=job_states)
hook.get_sync_hook = mock.AsyncMock(
return_value=mock.MagicMock(is_default_universe=mock.MagicMock(return_value=True))
)
hook.get_job_output = mock.AsyncMock(return_value={})
hook.get_records = mock.MagicMock(side_effect=lambda _: [[2]])
async def collect():
return [event.payload async for event in trigger.run()]
events = asyncio.run(collect())
return events[0], hook.get_job_status.await_count
def show(name, expected, event, polls):
got = f"{event['status']}: {event['message']}"
verdict = "OK " if got == expected else "BAD"
print(f" [{verdict}] {name}\n expected {expected!r}\n got
{got!r} ({polls} status calls)")
print("\n#74305 BigQuery check triggers")
value = BigQueryValueCheckTrigger(
conn_id="google_cloud_default", sql="SELECT COUNT(*) FROM t",
pass_value=2,
job_id="job_1", project_id="p", poll_interval=0.01,
)
event, polls = run_trigger(value, [RUNNING, DONE])
show("value check, job RUNNING on the first poll", "success: Job completed",
event, polls)
def interval():
return BigQueryIntervalCheckTrigger(
conn_id="google_cloud_default", first_job_id="job_1",
second_job_id="job_2", project_id="p",
table="t", metrics_thresholds={"COUNT(*)": 1.5}, poll_interval=0.01,
)
# get_job_status is awaited for the first job, then the second job, on every
poll.
event, polls = run_trigger(interval(), [RUNNING, DONE, DONE, DONE])
show("interval check, first job RUNNING on the first poll", "success: Job
completed", event, polls)
event, polls = run_trigger(interval(), [FAILED, RUNNING])
show("interval check, first job FAILED while the second runs", f"error:
{FAILED['message']}", event, polls)
print("\n#72953 GCSTimeSpanFileTransformOperator retry attempts")
from airflow.providers.google.cloud.operators.gcs import
GCSTimeSpanFileTransformOperator # noqa: E402
class FakeGCS:
"""A GCS client whose first download_failures downloads and
upload_failures uploads fail."""
def __init__(self, download_failures, upload_failures):
self.left = {"download": download_failures, "upload":
upload_failures}
self.calls = {"download": 0, "upload": 0}
self.uploaded = []
def _maybe_fail(self, kind):
self.calls[kind] += 1
if self.left[kind] > 0:
self.left[kind] -= 1
raise ServiceUnavailable(f"{kind} attempt {self.calls[kind]}:
503 backend unavailable")
def bucket(self, bucket_name):
return self
def blob(self, blob_name, chunk_size=None):
gcs = self
class Blob:
def download_to_filename(self, filename):
gcs._maybe_fail("download")
Path(filename).write_text("hello\n")
def upload_from_filename(self, filename):
gcs._maybe_fail("upload")
gcs.uploaded.append(blob_name)
return Blob()
def run_operator(download_failures, upload_failures, **attempts):
gcs = FakeGCS(download_failures, upload_failures)
hook = mock.MagicMock(project_id="p")
hook.list_by_timespan.return_value = ["in/2026-10-09/a.txt"]
hook.get_conn.return_value = gcs
with tempfile.TemporaryDirectory() as tmp:
script = Path(tmp) / "transform.sh"
script.write_text('#!/bin/sh\ncp -R "$1"/. "$2"/\n') # copy input
to output
script.chmod(script.stat().st_mode | stat.S_IEXEC)
op = GCSTimeSpanFileTransformOperator(
task_id="transform", source_bucket="src",
source_prefix="in/%Y-%m-%d/",
source_gcp_conn_id="google_cloud_default",
destination_bucket="dst",
destination_prefix="out/",
destination_gcp_conn_id="google_cloud_default",
transform_script=str(script), **attempts,
)
context = {
"data_interval_start": pendulum.datetime(2026, 10, 9),
"data_interval_end": pendulum.datetime(2026, 10, 10),
"ti": mock.MagicMock(),
}
with (
mock.patch("airflow.providers.google.cloud.operators.gcs.GCSHook",
return_value=hook),
mock.patch("tenacity.nap.time.sleep"), # keep the retry waits,
without waiting
):
try:
result = op.execute(context)
outcome = f"success, uploaded {result}"
except Exception as e:
outcome = f"{type(e).__name__}: {e}"
return outcome, gcs.calls
for name, failures, attempts, expect in [
("download fails twice, download_num_attempts=3", (2, 0),
{"download_num_attempts": 3}, "success"),
("upload fails once, upload_num_attempts=2", (0, 1),
{"upload_num_attempts": 2}, "success"),
("download fails 3 times, download_num_attempts=3", (3, 0),
{"download_num_attempts": 3}, "ServiceUnavailable"),
]:
outcome, calls = run_operator(*failures, **attempts)
verdict = "OK " if outcome.startswith(expect) else "BAD"
print(f" [{verdict}] {name}\n expected {expect}\n got
{outcome} "
f"(download calls {calls['download']}, upload calls
{calls['upload']})")
```
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
