shahar1 commented on code in PR #72030:
URL: https://github.com/apache/airflow/pull/72030#discussion_r4432215397
##########
providers/google/src/airflow/providers/google/cloud/operators/dataproc.py:
##########
@@ -2060,6 +2062,19 @@ def execute(self, context: Context):
raise AirflowException(f"Job failed:\n{job}")
if state == JobStatus.State.CANCELLED:
raise AirflowException(f"Job was cancelled:\n{job}")
+ execution_deadline = None
Review Comment:
Agreed, one poll interval of slack was not enough. The trigger now follows
the `KubernetesPodOperator` pattern and also bounds the poll itself:
- The deadline is checked at the top of the loop, before `get_job()`. A
trigger that wakes past the deadline cancels the job straight away.
- `get_job()` is wrapped in `asyncio.wait_for(..., timeout=remaining)`, so
its retries cannot run past the deadline.
- The polling sleep is capped at the deadline, so the trigger acts at D
instead of D + P.
- The framework backstop is now `remaining + max(60, 2 *
polling_interval_seconds)`, the same as KPO's `poll_buffer`.
The trigger now acts at D and has at least 60 s before the framework timeout
fires. `cancel_job()` still uses the client's default retry and runs inside
that window. Changes are in with tests for a slow `get_job()` and for a long
poll interval. I also ran `example_dataproc_spark_deferrable` end to end with
`execution_timeout` set on the submit task, and it passed.
--
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]
