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