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