EmrServerlessStartJobOperator does not raise airflow exception

[EzuA]

Hi! I am testing the Operator and I found that airflow is marking a task as SUCCESS even though the EMR job state is FAILED. I think the problem is in the EmrServerlessHook, this condition:

if state in failure_states:
    raise AirflowException(
        f"{object_type.title()} reached failure state {state}."
    )

should be at the end in the while loop.

Thanks!

[dacort]

Thanks @EzuA - I'll look into this today!

[fventuri-availity]

I think it is because this line (https://github.com/aws-samples/emr-serverless-samples/blob/main/airflow/emr_serverless/operators/emr.py#L220) has the 'desired_state' set to 'EmrServerlessJobSensor.TERMINAL_STATES', while I think it should have 'EmrServerlessJobSensor.SUCCESS_STATES':

                desired_state=EmrServerlessJobSensor.SUCCESS_STATES,

Franco

[dacort]

Yep, I think you're right, thank you!. :) Was just tracing the logic to verify.

[EzuA]

Yes, I have tested both options and they are working in my local environment. I think the option proposed by Franco is the best.

[dacort]

Testing this locally now as well - should have an updated release by end of day. Thanks to both of you!

[dacort]

Thanks again - fixed in v1.0.1.