You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "Leondon9 (via GitHub)" <gi...@apache.org> on 2026/09/25 10:39:08 UTC
[PR] Fix masked failure reason for deferrable EMR Serverless jobs [airflow]
Leondon9 opened a new pull request, #73719:
URL: https://github.com/apache/airflow/pull/73719
On a deferrable EMR Serverless job **failure or timeout**, the real failure reason was masked by a `KeyError`.
## The bug
`EmrServerlessStartJobTrigger` emits a success event carrying `job_details` (`{application_id, job_id}`), but a **failure** event carrying only `{status, message}`. `EmrServerlessStartJobOperator.execute_complete` then read `event["job_details"]["application_id"]` to cancel the run before raising — so on any real failure it raised `KeyError: 'job_details'`, hiding the actual reason. On resume `execute()` has not run, so `self.job_id` is also unset, meaning the cancel had no run id to act on even if `job_details` had been present.
Found while validating #73701 on live AWS: a genuine job failure surfaced only as `KeyError: 'job_details'` in the task logs, not the underlying Spark error.
## The fix
- The trigger now includes `job_details` in the **failure** event too, consistent with the success event.
- `execute_complete` reads the run id from the event (falling back to `self`) so it works on resume; cancels the run defensively (guarded, so an already-terminal run cannot mask the failure); and raises the **actual** failure message instead of a generic string.
This is an independent, pre-existing bug on `main`, unrelated to #73701 / #73702 — it was just noticed while validating them.
## Testing
- New tests: the trigger failure event carries `job_details`; `execute_complete` surfaces the real message; a failure event without `job_details` no longer raises `KeyError`; a cancel error does not mask the real reason.
- Full `test_emr_serverless.py` + EMR trigger suites green (92 passed).
- Provider `mypy` and other Docker-dependent prek hooks were not run locally (Docker unavailable); CI will validate them.
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes — Claude Code (Opus 4.8)
Generated-by: Claude Code (Opus 4.8) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix masked failure reason for deferrable EMR Serverless jobs [airflow]
Posted by "Leondon9 (via GitHub)" <gi...@apache.org>.
Leondon9 commented on PR #73719:
URL: https://github.com/apache/airflow/pull/73719#issuecomment-5892200629
> I have left some comments.
Thanks for the review, @SameerMesiah97 ! All comments addressed. Two done slightly differently:
- except: narrowed to (BotoCoreError, ClientError) (not just ClientError) so a connection error during the cancel can't re-mask the real reason.
- terminal failure vs. timeout: the trigger now tags the event with a failure_type (via WaiterTerminalFailure/WaiterMaxAttemptsError), and the operator raises an outcome-specific message and cancels only on a timeout — a terminal failure is already done, so it's no longer cancelled. Left the error case as don't cancel; happy to change if you prefer.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix masked failure reason for deferrable EMR Serverless jobs [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #73719:
URL: https://github.com/apache/airflow/pull/73719#discussion_r4108097130
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -1490,18 +1490,30 @@ def execute(self, context: Context, event: dict[str, Any] | None = None) -> str
return self.job_id
- def execute_complete(self, context: Context, event: dict[str, Any] | None = None) -> None:
+ def execute_complete(self, context: Context, event: dict[str, Any] | None = None) -> str | None:
validated_event = validate_execute_complete_event(event)
+ # On resume execute() has not run, so self.job_id is unset; read the run details the trigger
+ # sends in the event (present on both success and failure) and fall back to self as a guard.
+ job_details = validated_event.get("job_details") or {}
+ job_id = job_details.get("job_id") or self.job_id
+ application_id = job_details.get("application_id") or self.application_id
+
if validated_event["status"] == "success":
self.log.info("Serverless job completed")
- return validated_event["job_details"]["job_id"]
- self.log.info("Cancelling EMR Serverless job %s", self.job_id)
- self.hook.conn.cancel_job_run(
- applicationId=validated_event["job_details"]["application_id"],
- jobRunId=validated_event["job_details"]["job_id"],
+ return job_id
+ if job_id:
+ # The waiter fails on both a job failure and a poll timeout; on timeout the run may still
+ # be active, so try to cancel it to avoid an orphan. Guard it so a cancel error (e.g. an
+ # already-terminal run) does not mask the real failure reason surfaced below.
+ self.log.info("Cancelling EMR Serverless job %s after failure or timeout", job_id)
+ try:
+ self.hook.conn.cancel_job_run(applicationId=application_id, jobRunId=job_id)
+ except Exception:
Review Comment:
Could we narrow this `Exception `to the errors that are expected to arise?
##########
providers/amazon/tests/unit/amazon/aws/operators/test_emr_serverless.py:
##########
@@ -693,6 +693,58 @@ def test_execute_complete_deferrable_failure_triggers_cancel(
log_msgs = [call.args[0] for call in mock_log.call_args_list]
assert any("Cancelling EMR Serverless job" in msg for msg in log_msgs)
+ @mock.patch.object(EmrServerlessHook, "conn")
+ def test_execute_complete_failure_surfaces_message(self, mock_conn):
+ operator = EmrServerlessStartJobOperator(
+ task_id="test_task",
+ application_id="app-id",
+ execution_role_arn="arn",
+ job_driver={"sparkSubmit": {"entryPoint": "s3://x"}},
+ deferrable=True,
+ )
+ failed_event = {
+ "status": "failure",
+ "message": "Serverless Job failed: boom",
+ "job_details": {"application_id": "app-id", "job_id": "job-id"},
+ }
+ with pytest.raises(AirflowException, match="boom"):
+ operator.execute_complete(mock.MagicMock(), failed_event)
+
+ @mock.patch.object(EmrServerlessHook, "conn")
+ def test_execute_complete_failure_without_job_details_does_not_keyerror(self, mock_conn):
+ """A trigger failure event without job_details must surface the reason, not raise KeyError."""
+ operator = EmrServerlessStartJobOperator(
+ task_id="test_task",
+ application_id="app-id",
+ execution_role_arn="arn",
+ job_driver={"sparkSubmit": {"entryPoint": "s3://x"}},
+ deferrable=True,
+ )
+ failed_event = {"status": "failure", "message": "Serverless Job failed: boom"}
+ with pytest.raises(AirflowException, match="boom"):
+ operator.execute_complete(mock.MagicMock(), failed_event)
+ # No run id available on resume, so nothing to cancel.
+ mock_conn.cancel_job_run.assert_not_called()
+
+ @mock.patch.object(EmrServerlessHook, "conn")
+ def test_execute_complete_failure_cancel_error_does_not_mask_reason(self, mock_conn):
+ mock_conn.cancel_job_run.side_effect = Exception("run is not in a cancellable state")
+ operator = EmrServerlessStartJobOperator(
+ task_id="test_task",
+ application_id="app-id",
+ execution_role_arn="arn",
+ job_driver={"sparkSubmit": {"entryPoint": "s3://x"}},
+ deferrable=True,
+ )
+ failed_event = {
+ "status": "failure",
+ "message": "Serverless Job failed: boom",
+ "job_details": {"application_id": "app-id", "job_id": "job-id"},
+ }
+ with pytest.raises(AirflowException, match="boom"):
+ operator.execute_complete(mock.MagicMock(), failed_event)
+ mock_conn.cancel_job_run.assert_called_once()
Review Comment:
I would assert the cancellation arguments too. Please see the below:
```
mock_conn.cancel_job_run.assert_called_once_with(
applicationId="app-id", jobRunId="job-id"
)
```
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -1490,18 +1490,30 @@ def execute(self, context: Context, event: dict[str, Any] | None = None) -> str
return self.job_id
- def execute_complete(self, context: Context, event: dict[str, Any] | None = None) -> None:
+ def execute_complete(self, context: Context, event: dict[str, Any] | None = None) -> str | None:
validated_event = validate_execute_complete_event(event)
+ # On resume execute() has not run, so self.job_id is unset; read the run details the trigger
+ # sends in the event (present on both success and failure) and fall back to self as a guard.
Review Comment:
Comment is a bit unclear. The below is more clear:
`#After deferral, self.job_id may be unavailable. Use the IDs supplied by the trigger event for cancellation`
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -1490,18 +1490,30 @@ def execute(self, context: Context, event: dict[str, Any] | None = None) -> str
return self.job_id
- def execute_complete(self, context: Context, event: dict[str, Any] | None = None) -> None:
+ def execute_complete(self, context: Context, event: dict[str, Any] | None = None) -> str | None:
validated_event = validate_execute_complete_event(event)
+ # On resume execute() has not run, so self.job_id is unset; read the run details the trigger
+ # sends in the event (present on both success and failure) and fall back to self as a guard.
+ job_details = validated_event.get("job_details") or {}
+ job_id = job_details.get("job_id") or self.job_id
+ application_id = job_details.get("application_id") or self.application_id
+
if validated_event["status"] == "success":
self.log.info("Serverless job completed")
- return validated_event["job_details"]["job_id"]
- self.log.info("Cancelling EMR Serverless job %s", self.job_id)
- self.hook.conn.cancel_job_run(
- applicationId=validated_event["job_details"]["application_id"],
- jobRunId=validated_event["job_details"]["job_id"],
+ return job_id
+ if job_id:
+ # The waiter fails on both a job failure and a poll timeout; on timeout the run may still
+ # be active, so try to cancel it to avoid an orphan. Guard it so a cancel error (e.g. an
+ # already-terminal run) does not mask the real failure reason surfaced below.
+ self.log.info("Cancelling EMR Serverless job %s after failure or timeout", job_id)
+ try:
+ self.hook.conn.cancel_job_run(applicationId=application_id, jobRunId=job_id)
+ except Exception:
+ self.log.exception("Failed to cancel EMR Serverless job %s", job_id)
+ raise AirflowException(
+ f"EMR Serverless job failed or timed out in deferrable mode: {validated_event.get('message', '')}"
)
Review Comment:
1) `AirflowException` is to avoided in provider code. Perhaps, you can use `RuntimeError` instead.
2) Could we distinguish a terminal job failure from exhausted polling attempts in the trigger event, then raise a message specific to the outcome here? At present both arrive as `"failure"`, so “failed or timed out” leaves users unable to tell what happened.
##########
providers/amazon/tests/unit/amazon/aws/operators/test_emr_serverless.py:
##########
@@ -693,6 +693,58 @@ def test_execute_complete_deferrable_failure_triggers_cancel(
log_msgs = [call.args[0] for call in mock_log.call_args_list]
assert any("Cancelling EMR Serverless job" in msg for msg in log_msgs)
+ @mock.patch.object(EmrServerlessHook, "conn")
+ def test_execute_complete_failure_surfaces_message(self, mock_conn):
+ operator = EmrServerlessStartJobOperator(
+ task_id="test_task",
+ application_id="app-id",
+ execution_role_arn="arn",
+ job_driver={"sparkSubmit": {"entryPoint": "s3://x"}},
+ deferrable=True,
+ )
+ failed_event = {
+ "status": "failure",
+ "message": "Serverless Job failed: boom",
+ "job_details": {"application_id": "app-id", "job_id": "job-id"},
+ }
+ with pytest.raises(AirflowException, match="boom"):
+ operator.execute_complete(mock.MagicMock(), failed_event)
+
+ @mock.patch.object(EmrServerlessHook, "conn")
+ def test_execute_complete_failure_without_job_details_does_not_keyerror(self, mock_conn):
+ """A trigger failure event without job_details must surface the reason, not raise KeyError."""
+ operator = EmrServerlessStartJobOperator(
+ task_id="test_task",
+ application_id="app-id",
+ execution_role_arn="arn",
+ job_driver={"sparkSubmit": {"entryPoint": "s3://x"}},
+ deferrable=True,
+ )
+ failed_event = {"status": "failure", "message": "Serverless Job failed: boom"}
+ with pytest.raises(AirflowException, match="boom"):
+ operator.execute_complete(mock.MagicMock(), failed_event)
+ # No run id available on resume, so nothing to cancel.
+ mock_conn.cancel_job_run.assert_not_called()
+
+ @mock.patch.object(EmrServerlessHook, "conn")
+ def test_execute_complete_failure_cancel_error_does_not_mask_reason(self, mock_conn):
+ mock_conn.cancel_job_run.side_effect = Exception("run is not in a cancellable state")
+ operator = EmrServerlessStartJobOperator(
+ task_id="test_task",
+ application_id="app-id",
+ execution_role_arn="arn",
+ job_driver={"sparkSubmit": {"entryPoint": "s3://x"}},
+ deferrable=True,
+ )
+ failed_event = {
+ "status": "failure",
+ "message": "Serverless Job failed: boom",
+ "job_details": {"application_id": "app-id", "job_id": "job-id"},
+ }
+ with pytest.raises(AirflowException, match="boom"):
Review Comment:
Match the full raised message, or at least the meaningful failure reason, rather than just "boom". This feedback applies to all applicable tests.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org