VladaZakharova commented on code in PR #72030:
URL: https://github.com/apache/airflow/pull/72030#discussion_r4026803314
##########
providers/google/src/airflow/providers/google/cloud/operators/dataproc.py:
##########
@@ -2089,6 +2106,10 @@ def execute_complete(self, context, event=None) -> None:
"""
job_state = event["job_state"]
job_id = event["job_id"]
+ if job_state == "TIMED_OUT":
+ raise AirflowException(
Review Comment:
In community now we are trying to reduce the number of AirflowExceptions, so
can youplease chage it to the more meaningful?
##########
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:
I think the one-poll-interval backstop can still win and leave the Dataproc
job running.
Consider a poll immediately before deadline D. The trigger sees RUNNING,
sleeps for P, and wakes around D + P—exactly when the framework timeout
expires. It still needs to call get_job(), call cancel_job(), publish the
trigger event, have Airflow process that event.
Scheduler can therefore take the framework timeout path first, recreating
the orphaned-job problem this change is meant to solve. The risk is larger
because the deadline is checked only after get_job(), whose retries are not
bounded by the remaining execution time.
You can follow KubernetesPodOperator example (trigger_timeout parameter).
##########
providers/google/tests/unit/google/cloud/operators/test_dataproc.py:
##########
@@ -1867,6 +1867,48 @@ def test_execute_deferrable(self, mock_trigger_hook,
mock_hook):
assert isinstance(exc.value.trigger, DataprocSubmitTrigger)
assert exc.value.method_name == GOOGLE_DEFAULT_DEFERRABLE_METHOD_NAME
+ @mock.patch(DATAPROC_PATH.format("DataprocHook"))
+ @mock.patch(DATAPROC_TRIGGERS_PATH.format("DataprocAsyncHook"))
+ def test_execute_deferrable_with_execution_timeout(self,
mock_trigger_hook, mock_hook):
+ mock_hook.return_value.submit_job.return_value.reference.job_id =
TEST_JOB_ID
+
+ op = DataprocSubmitJobOperator(
+ task_id=TASK_ID,
+ region=GCP_REGION,
+ project_id=GCP_PROJECT,
+ job={},
+ gcp_conn_id=GCP_CONN_ID,
+ deferrable=True,
+ execution_timeout=dt.timedelta(hours=1),
+ )
+ start_date = dt.datetime(2026, 1, 1, tzinfo=dt.timezone.utc)
Review Comment:
The operator test uses January 1, 2026 without freezing time. That date was
already in the past when the PR was opened, so max(remaining, 0) always becomes
zero. It only asserts that the timeout is non-None.
The tests should use time_machine and assert the exact timeout.
--
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]
