o-nikolas commented on code in PR #73701:
URL: https://github.com/apache/airflow/pull/73701#discussion_r4354094505
##########
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:
This guard runs on every Airflow version, but durable only works on 3.3+.
Below 3.3, durable=True with deferrable=True does no reconnect, yet still turns
cancel_on_kill off. The trigger's CancelledError handler still uses
cancel_on_kill there, so killing the task leaves the job running. Could you
gate this on AIRFLOW_V_3_3_PLUS, with a test for the older case?
##########
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:
nit: if the stored run already succeeded, this still starts the application
when it has already stopped. Could you check the stored id before the
application start, and return right away on success?
##########
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:
With the __init__ guard, a durable deferrable task never reaches this code,
so the docstring's reason no longer holds. With durable=False the next try
submits a fresh run anyway, so this only makes on_kill slower (up to ~28s). It
also stops at CANCELLING, which hasn't settled yet. I'd drop this from the PR,
or move it to its own PR with its own reason.
--
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]
