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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
