Leondon9 commented on code in PR #73701:
URL: https://github.com/apache/airflow/pull/73701#discussion_r4369873497
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -1369,7 +1386,24 @@ def __init__(
self.job_id: str | None = None
self.deferrable = deferrable
self.enable_application_ui_links = enable_application_ui_links
- self.cancel_on_kill = cancel_on_kill
+ # In deferrable mode durable reconnects to the still-running job on
clear, while cancel_on_kill
+ # (which only applies to the deferrable trigger) kills it: the two are
mutually exclusive there.
+ # Because the worker re-executes before the trigger's on_kill runs, a
durable retry would
+ # reconnect to a run that on_kill is about to cancel and then fail, so
reject the contradiction
+ # and keep cancel_on_kill off when durable is enabled. Synchronous
runs have no such race and
+ # cancel_on_kill does not apply to them, so the value is left
untouched there.
+ if durable and deferrable:
Review Comment:
Gated on `AIRFLOW_V_3_3_PLUS`, with a test for both sides. I also dropped
the `ValueError` for `durable=True` with `cancel_on_kill=True`. With the gate
it would only fire on 3.3+, so a Dag that parses on 3.2 would fail to import
after an upgrade. #73702 (Glue) handles the same combination by stopping the
previous run and submitting a fresh one, so EMR now does the same: with
`cancel_on_kill=True` set explicitly, a retry cancels the stored run if it is
still active and submits fresh instead of reconnecting. Leaving
`cancel_on_kill` unset still turns it off, so reconnect stays what `durable`
does by default. Happy to split this out if you'd rather keep this PR to its
original scope.
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -1409,37 +1443,11 @@ def execute(self, context: Context, event: dict[str,
Any] | None = None) -> str
self.log.info("Starting job on Application: %s", self.application_id)
self.name = self.name or self.config.pop("name",
f"emr_serverless_job_airflow_{uuid4()}")
- configuration_overrides = self.configuration_overrides
- if self.openlineage_inject_parent_job_info:
- self.log.info("Injecting OpenLineage parent job information into
EMR Serverless configuration.")
- configuration_overrides =
inject_parent_job_information_into_emr_serverless_properties(
- configuration_overrides, context
- )
- if self.openlineage_inject_transport_info:
- self.log.info("Injecting OpenLineage transport information into
EMR Serverless configuration.")
- configuration_overrides =
inject_transport_information_into_emr_serverless_properties(
- configuration_overrides, context
- )
-
- args = {
- "clientToken": self.client_request_token,
- "applicationId": self.application_id,
- "executionRoleArn": self.execution_role_arn,
- "jobDriver": self.job_driver,
- "name": self.name,
- **self.config,
- }
- if configuration_overrides is not None:
- args["configurationOverrides"] = configuration_overrides
- response = self.hook.conn.start_job_run(
- **args,
- )
-
- if response["ResponseMetadata"]["HTTPStatusCode"] != 200:
- raise AirflowException(f"EMR serverless job failed to start:
{response}")
-
- self.job_id = response["jobRunId"]
- self.log.info("EMR serverless job started: %s", self.job_id)
+ self.job_id = self._reconnect_to_existing_job_run(context) if
self.durable else None
Review Comment:
Done. The stored run is now checked before the application start, and a run
that already succeeded returns right away after persisting the links.
##########
providers/amazon/src/airflow/providers/amazon/aws/triggers/emr.py:
##########
@@ -706,17 +711,45 @@ async def on_kill(self) -> None:
This hook is available in Airflow 3.3+ via BaseTrigger.on_kill().
For older Airflow versions, the CancelledError handler in run()
provides
the same cancellation behavior.
+
+ After requesting the cancellation, wait until the run leaves its
active states. A durable
+ retry reconnects to whatever run id is stored, so returning while the
run still looks active
+ would let the retry reconnect to a run that is about to be cancelled
and then fail. This
+ mirrors the synchronous ``EmrServerlessStartJobOperator.on_kill``,
which also waits for a
+ terminal state.
"""
if self.job_id and self.cancel_on_kill:
self.log.info(
"Cancelling EMR Serverless job. Application ID: %s, Job ID:
%s",
self.application_id,
self.job_id,
)
- await sync_to_async(self.hook().conn.cancel_job_run)(
- applicationId=self.application_id, jobRunId=self.job_id
- )
- self.log.info("EMR Serverless job %s cancelled successfully.",
self.job_id)
+ conn = self.hook().conn
+ await
sync_to_async(conn.cancel_job_run)(applicationId=self.application_id,
jobRunId=self.job_id)
+ await self._wait_until_not_active(conn)
Review Comment:
Dropped. The trigger is back to what's on main.
--
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]
