You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "syedahsn (via GitHub)" <gi...@apache.org> on 2023/06/26 18:17:30 UTC
[GitHub] [airflow] syedahsn opened a new pull request, #32151: Get failure information on EMR job failure
syedahsn opened a new pull request, #32151:
URL: https://github.com/apache/airflow/pull/32151
Currently, the logs do not give any information about the progress of an EMR job that has been scheduled. This PR uses the new `wait` util method to display the status of the EMR jobs, and if a job fails, it calls the `describe_step` method to log the reason for the job failure.
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of an existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#pull-request-guidelines)** for more information.
In case of fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in a newsfragment file, named `{pr_number}.significant.rst` or `{issue_number}.significant.rst`, in [newsfragments](https://github.com/apache/airflow/tree/main/newsfragments).
--
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
[GitHub] [airflow] o-nikolas commented on a diff in pull request #32151: Get failure information on EMR job failure
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on code in PR #32151:
URL: https://github.com/apache/airflow/pull/32151#discussion_r1284891154
##########
airflow/providers/amazon/aws/hooks/emr.py:
##########
@@ -169,16 +172,23 @@ def add_job_flow_steps(
if wait_for_completion:
waiter = self.get_conn().get_waiter("step_complete")
for step_id in response["StepIds"]:
- waiter.wait(
- ClusterId=job_flow_id,
- StepId=step_id,
- WaiterConfig=prune_dict(
- {
- "Delay": waiter_delay,
- "MaxAttempts": waiter_max_attempts,
- }
- ),
- )
+ try:
+ wait(
+ waiter=waiter,
+ waiter_max_attempts=waiter_max_attempts,
+ waiter_delay=waiter_delay,
+ args={"ClusterId": job_flow_id, "StepId": step_id},
+ failure_message=f"EMR Steps failed: {step_id}",
+ status_message="EMR Step status is",
+ status_args=["Step.Status.State", "Step.Status.StateChangeReason"],
+ )
+ except AirflowException as ex:
+ if "EMR Steps failed" in str(ex):
+ resp = self.get_conn().describe_step(ClusterId=job_flow_id, StepId=step_id)
+ failure_details = resp["Step"]["Status"].get("FailureDetails", None)
+ if failure_details:
+ self.log.error("EMR Steps failed: %s", failure_details)
+ raise
Review Comment:
It's hard to tell in the diff view, but should this raise be at the scope of the except? Right now we only re-raise if "EMR Steps failed" is in the exception message, so we wont re-raise otherwise. But I think that conditional should only dictate whether we request a describe for failure details.
But I may have that wrong, let me know if the raise guarded by the if was intentional.
--
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
[GitHub] [airflow] dacort commented on a diff in pull request #32151: Get failure information on EMR job failure
Posted by "dacort (via GitHub)" <gi...@apache.org>.
dacort commented on code in PR #32151:
URL: https://github.com/apache/airflow/pull/32151#discussion_r1267061309
##########
airflow/providers/amazon/aws/hooks/emr.py:
##########
@@ -169,16 +172,21 @@ def add_job_flow_steps(
if wait_for_completion:
waiter = self.get_conn().get_waiter("step_complete")
for step_id in response["StepIds"]:
- waiter.wait(
- ClusterId=job_flow_id,
- StepId=step_id,
- WaiterConfig=prune_dict(
- {
- "Delay": waiter_delay,
- "MaxAttempts": waiter_max_attempts,
- }
- ),
- )
+ try:
+ wait(
+ waiter=waiter,
+ max_attempts=waiter_max_attempts,
+ waiter_delay=waiter_delay,
+ args={"ClusterId": job_flow_id, "StepId": step_id},
+ failure_message=f"EMR Steps failed: {step_id}",
+ status_message="EMR Step status is",
+ status_args=["Step.Status.State", "Step.Status.StateChangeReason"],
+ )
+ except AirflowException as ex:
+ if "EMR Steps failed" in str(ex):
+ resp = self.get_conn().describe_step(ClusterId=job_flow_id, StepId=step_id)
+ self.log.error("EMR Steps failed: %s", resp["Step"]["Status"]["StateChangeReason"])
Review Comment:
@o-nikolas `FailureDetails` should surface common reasons that a step failed and it will do so before the logs get synced to S3. I'm not as familiar with `StageChangedReason`, but I'm not sure it will have potential details about _why_ the job failed. I think it's more related to service-specific reasons why a step state changed (e.g. somebody cancelled it). Here's an example of a job I ran with an invalid entrypoint file. `FailureDetails` is populated immediately and the log file it shows didn't sync to S3 until 4 minutes after the step failed.
```json
{
"Step": {
"Id": "s-00928142VC3M3WYAG8XN",
"Name": "test-fail",
"Config": {
"Jar": "command-runner.jar",
"Properties": {},
"Args": [
"spark-submit",
"--deploy-mode",
"cluster",
"/tmp/fail.py"
]
},
"ActionOnFailure": "CONTINUE",
"Status": {
"State": "FAILED",
"StateChangeReason": {},
"FailureDetails": {
"Reason": "The given path is invalid.",
"Message": "Exception in thread \"main\" java.io.FileNotFoundException: File file:/mnt/tmp/fail.py does not exist",
"LogFile": "s3://aws-logs-568026268536-us-west-2/elasticmapreduce/j-2WPQL41AO1169/steps/s-00928142VC3M3WYAG8XN/stderr.gz"
},
"Timeline": {
"CreationDateTime": "2023-07-18T09:41:40.728000-07:00",
"StartDateTime": "2023-07-18T09:41:57.422000-07:00",
"EndDateTime": "2023-07-18T09:42:03.499000-07:00"
}
}
}
}
```
And here's one where I cancelled the job before it ran with `StateChangeReason` populated.
```json
{
"Step": {
"Id": "s-07516842NY2M4IHHH71T",
"Name": "test-fail",
"Config": {
"Jar": "command-runner.jar",
"Properties": {},
"Args": [
"spark-submit",
"--deploy-mode",
"cluster",
"/tmp/fail.py"
]
},
"ActionOnFailure": "CONTINUE",
"Status": {
"State": "CANCELLED",
"StateChangeReason": {
"Message": "Cancelled by user"
},
"Timeline": {
"CreationDateTime": "2023-07-18T09:48:45.141000-07:00"
}
}
}
}
```
--
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
[GitHub] [airflow] o-nikolas merged pull request #32151: Get failure information on EMR job failure
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas merged PR #32151:
URL: https://github.com/apache/airflow/pull/32151
--
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
[GitHub] [airflow] o-nikolas commented on a diff in pull request #32151: Get failure information on EMR job failure
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on code in PR #32151:
URL: https://github.com/apache/airflow/pull/32151#discussion_r1244180804
##########
airflow/providers/amazon/aws/hooks/emr.py:
##########
@@ -169,16 +172,21 @@ def add_job_flow_steps(
if wait_for_completion:
waiter = self.get_conn().get_waiter("step_complete")
for step_id in response["StepIds"]:
- waiter.wait(
- ClusterId=job_flow_id,
- StepId=step_id,
- WaiterConfig=prune_dict(
- {
- "Delay": waiter_delay,
- "MaxAttempts": waiter_max_attempts,
- }
- ),
- )
+ try:
+ wait(
+ waiter=waiter,
+ max_attempts=waiter_max_attempts,
+ waiter_delay=waiter_delay,
+ args={"ClusterId": job_flow_id, "StepId": step_id},
+ failure_message=f"EMR Steps failed: {step_id}",
+ status_message="EMR Step status is",
+ status_args=["Step.Status.State", "Step.Status.StateChangeReason"],
+ )
+ except AirflowException as ex:
+ if "EMR Steps failed" in str(ex):
+ resp = self.get_conn().describe_step(ClusterId=job_flow_id, StepId=step_id)
+ self.log.error("EMR Steps failed: %s", resp["Step"]["Status"]["StateChangeReason"])
Review Comment:
Are there any other portions of the status that are worth logging in addition to the state change 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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] syedahsn commented on a diff in pull request #32151: Get failure information on EMR job failure
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #32151:
URL: https://github.com/apache/airflow/pull/32151#discussion_r1244499448
##########
airflow/providers/amazon/aws/hooks/emr.py:
##########
@@ -169,16 +172,21 @@ def add_job_flow_steps(
if wait_for_completion:
waiter = self.get_conn().get_waiter("step_complete")
for step_id in response["StepIds"]:
- waiter.wait(
- ClusterId=job_flow_id,
- StepId=step_id,
- WaiterConfig=prune_dict(
- {
- "Delay": waiter_delay,
- "MaxAttempts": waiter_max_attempts,
- }
- ),
- )
+ try:
+ wait(
+ waiter=waiter,
+ max_attempts=waiter_max_attempts,
+ waiter_delay=waiter_delay,
+ args={"ClusterId": job_flow_id, "StepId": step_id},
+ failure_message=f"EMR Steps failed: {step_id}",
+ status_message="EMR Step status is",
+ status_args=["Step.Status.State", "Step.Status.StateChangeReason"],
+ )
+ except AirflowException as ex:
+ if "EMR Steps failed" in str(ex):
+ resp = self.get_conn().describe_step(ClusterId=job_flow_id, StepId=step_id)
+ self.log.error("EMR Steps failed: %s", resp["Step"]["Status"]["StateChangeReason"])
Review Comment:
Not really. In the documentation, there is supposed to be a `FailureDetails` field that should provide details, but I haven't seen that in the actual response body.
--
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
[GitHub] [airflow] syedahsn commented on pull request #32151: Get failure information on EMR job failure
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on PR #32151:
URL: https://github.com/apache/airflow/pull/32151#issuecomment-1608519851
The error messages might not explain the root cause of the problem, but they will at least be a starting point for an investigation. This is an example where I manually deleted the cluster before the job was run:
```
[2023-06-26, 23:47:57 UTC] {emr.py:146} INFO - Adding steps to j-2D6XS5VTI1WGZ
[2023-06-26, 23:47:57 UTC] {emr.py:171} INFO - Steps ['s-08516095GAHC1SG5B4N'] added to JobFlow
[2023-06-26, 23:47:57 UTC] {waiter_with_logging.py:74} INFO - EMR Step status is: PENDING - {}
[2023-06-26, 23:48:36 UTC] {emr.py:188} ERROR - EMR Steps failed: {'Message': 'Job terminated'}
[2023-06-26, 23:48:36 UTC] {taskinstance.py:1916} ERROR - Task failed with exception
Traceback (most recent call last):
File "/opt/airflow/airflow/providers/amazon/aws/utils/waiter_with_logging.py", line 68, in wait
waiter.wait(**args, WaiterConfig={"MaxAttempts": 1})
File "/usr/local/lib/python3.8/site-packages/botocore/waiter.py", line 55, in wait
Waiter.wait(self, **kwargs)
File "/usr/local/lib/python3.8/site-packages/botocore/waiter.py", line 375, in wait
raise WaiterError(
botocore.exceptions.WaiterError: Waiter StepComplete failed: Waiter encountered a terminal failure state: For expression "Step.Status.State" we matched expected path: "CANCELLED"
```
This message is logged `EMR Steps failed: {'Message': 'Job terminated'}`, which provides some indication of what happened.
If a job fails due to some issue with the job itself, then, like you said, that information will not be logged here.
I agree its not much, but its better than nothing. What do you think?
--
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
[GitHub] [airflow] o-nikolas commented on a diff in pull request #32151: Get failure information on EMR job failure
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on code in PR #32151:
URL: https://github.com/apache/airflow/pull/32151#discussion_r1265684845
##########
airflow/providers/amazon/aws/hooks/emr.py:
##########
@@ -169,16 +172,21 @@ def add_job_flow_steps(
if wait_for_completion:
waiter = self.get_conn().get_waiter("step_complete")
for step_id in response["StepIds"]:
- waiter.wait(
- ClusterId=job_flow_id,
- StepId=step_id,
- WaiterConfig=prune_dict(
- {
- "Delay": waiter_delay,
- "MaxAttempts": waiter_max_attempts,
- }
- ),
- )
+ try:
+ wait(
+ waiter=waiter,
+ max_attempts=waiter_max_attempts,
+ waiter_delay=waiter_delay,
+ args={"ClusterId": job_flow_id, "StepId": step_id},
+ failure_message=f"EMR Steps failed: {step_id}",
+ status_message="EMR Step status is",
+ status_args=["Step.Status.State", "Step.Status.StateChangeReason"],
+ )
+ except AirflowException as ex:
+ if "EMR Steps failed" in str(ex):
+ resp = self.get_conn().describe_step(ClusterId=job_flow_id, StepId=step_id)
+ self.log.error("EMR Steps failed: %s", resp["Step"]["Status"]["StateChangeReason"])
Review Comment:
Is there any other source for useful information you know of, @dacort?
--
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
[GitHub] [airflow] eladkal commented on pull request #32151: Get failure information on EMR job failure
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on PR #32151:
URL: https://github.com/apache/airflow/pull/32151#issuecomment-1608009428
> and if a job fails, it calls the `describe_step` method to log the reason for the job failure.
My experience with EMR is that it takes time for logs to be populated. More often the failure details are actually in stdout rather than stderr and the general error message is not very helpful.
I'm curious to what exactly is being registered here. Is it possibly yo share example?
--
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
[GitHub] [airflow] eladkal commented on pull request #32151: Get failure information on EMR job failure
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on PR #32151:
URL: https://github.com/apache/airflow/pull/32151#issuecomment-1608913983
> If a job fails due to some issue with the job itself, then, like you said, that information will not be logged here.
> I agree its not much, but its better than nothing. What do you think?
Since Airflow task failed you already know everything the step encountered a terminal state.
Maybe worth consulting with @dacort to see if somehow we can bring a more informative message?
Either way I'm OK with it as is.. was just curious to see what is logged in the message.
--
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
[GitHub] [airflow] eladkal commented on a diff in pull request #32151: Get failure information on EMR job failure
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on code in PR #32151:
URL: https://github.com/apache/airflow/pull/32151#discussion_r1244547924
##########
airflow/providers/amazon/aws/hooks/emr.py:
##########
@@ -169,16 +172,21 @@ def add_job_flow_steps(
if wait_for_completion:
waiter = self.get_conn().get_waiter("step_complete")
for step_id in response["StepIds"]:
- waiter.wait(
- ClusterId=job_flow_id,
- StepId=step_id,
- WaiterConfig=prune_dict(
- {
- "Delay": waiter_delay,
- "MaxAttempts": waiter_max_attempts,
- }
- ),
- )
+ try:
+ wait(
+ waiter=waiter,
+ max_attempts=waiter_max_attempts,
+ waiter_delay=waiter_delay,
+ args={"ClusterId": job_flow_id, "StepId": step_id},
+ failure_message=f"EMR Steps failed: {step_id}",
+ status_message="EMR Step status is",
+ status_args=["Step.Status.State", "Step.Status.StateChangeReason"],
+ )
+ except AirflowException as ex:
+ if "EMR Steps failed" in str(ex):
+ resp = self.get_conn().describe_step(ClusterId=job_flow_id, StepId=step_id)
+ self.log.error("EMR Steps failed: %s", resp["Step"]["Status"]["StateChangeReason"])
Review Comment:
When I have a failed job it takes 3-4 minutes before the logs appear and the issue is discovered. I wonder if the empty body reflects this as querying too soon when there is not yet information from the logs.
--
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
[GitHub] [airflow] vandonr-amz commented on a diff in pull request #32151: Get failure information on EMR job failure
Posted by "vandonr-amz (via GitHub)" <gi...@apache.org>.
vandonr-amz commented on code in PR #32151:
URL: https://github.com/apache/airflow/pull/32151#discussion_r1245750324
##########
airflow/providers/amazon/aws/hooks/emr.py:
##########
@@ -169,16 +172,21 @@ def add_job_flow_steps(
if wait_for_completion:
waiter = self.get_conn().get_waiter("step_complete")
for step_id in response["StepIds"]:
- waiter.wait(
- ClusterId=job_flow_id,
- StepId=step_id,
- WaiterConfig=prune_dict(
- {
- "Delay": waiter_delay,
- "MaxAttempts": waiter_max_attempts,
- }
- ),
- )
+ try:
+ wait(
+ waiter=waiter,
+ max_attempts=waiter_max_attempts,
+ waiter_delay=waiter_delay,
+ args={"ClusterId": job_flow_id, "StepId": step_id},
+ failure_message=f"EMR Steps failed: {step_id}",
+ status_message="EMR Step status is",
+ status_args=["Step.Status.State", "Step.Status.StateChangeReason"],
+ )
+ except AirflowException as ex:
+ if "EMR Steps failed" in str(ex):
+ resp = self.get_conn().describe_step(ClusterId=job_flow_id, StepId=step_id)
+ self.log.error("EMR Steps failed: %s", resp["Step"]["Status"]["StateChangeReason"])
Review Comment:
the waiter is already querying this endpoint, so you're going to get the same info as what's already printed by the method since you put `"Step.Status.StateChangeReason"` in the status args.
+1 to use the `wait` method, but -1 to this `except`
--
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
[GitHub] [airflow] o-nikolas commented on a diff in pull request #32151: Get failure information on EMR job failure
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on code in PR #32151:
URL: https://github.com/apache/airflow/pull/32151#discussion_r1283993039
##########
airflow/providers/amazon/aws/hooks/emr.py:
##########
@@ -169,16 +172,23 @@ def add_job_flow_steps(
if wait_for_completion:
waiter = self.get_conn().get_waiter("step_complete")
for step_id in response["StepIds"]:
- waiter.wait(
- ClusterId=job_flow_id,
- StepId=step_id,
- WaiterConfig=prune_dict(
- {
- "Delay": waiter_delay,
- "MaxAttempts": waiter_max_attempts,
- }
- ),
- )
+ try:
+ wait(
+ waiter=waiter,
+ waiter_max_attempts=waiter_max_attempts,
+ waiter_delay=waiter_delay,
+ args={"ClusterId": job_flow_id, "StepId": step_id},
+ failure_message=f"EMR Steps failed: {step_id}",
+ status_message="EMR Step status is",
+ status_args=["Step.Status.State", "Step.Status.StateChangeReason"],
+ )
+ except AirflowException as ex:
+ if "EMR Steps failed" in str(ex):
+ resp = self.get_conn().describe_step(ClusterId=job_flow_id, StepId=step_id)
+ self.log.error(
+ "EMR Steps failed: %s", resp["Step"]["Status"].get("FailureDetails", None)
Review Comment:
Maybe check if you have any failure details first and don't log at all if you don't? As of now you'll get a weird message that looks like it should have more details but people may get nothing there.
--
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
[GitHub] [airflow] syedahsn commented on a diff in pull request #32151: Get failure information on EMR job failure
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #32151:
URL: https://github.com/apache/airflow/pull/32151#discussion_r1284914301
##########
airflow/providers/amazon/aws/hooks/emr.py:
##########
@@ -169,16 +172,23 @@ def add_job_flow_steps(
if wait_for_completion:
waiter = self.get_conn().get_waiter("step_complete")
for step_id in response["StepIds"]:
- waiter.wait(
- ClusterId=job_flow_id,
- StepId=step_id,
- WaiterConfig=prune_dict(
- {
- "Delay": waiter_delay,
- "MaxAttempts": waiter_max_attempts,
- }
- ),
- )
+ try:
+ wait(
+ waiter=waiter,
+ waiter_max_attempts=waiter_max_attempts,
+ waiter_delay=waiter_delay,
+ args={"ClusterId": job_flow_id, "StepId": step_id},
+ failure_message=f"EMR Steps failed: {step_id}",
+ status_message="EMR Step status is",
+ status_args=["Step.Status.State", "Step.Status.StateChangeReason"],
+ )
+ except AirflowException as ex:
+ if "EMR Steps failed" in str(ex):
+ resp = self.get_conn().describe_step(ClusterId=job_flow_id, StepId=step_id)
+ failure_details = resp["Step"]["Status"].get("FailureDetails", None)
+ if failure_details:
+ self.log.error("EMR Steps failed: %s", failure_details)
+ raise
Review Comment:
You're right that makes sense. I moved it out of the `if` block.
--
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