EmrCreateJobFlowOperator should not wait_for_completion if wait_for_completion=false deferrable=true

Apache Airflow version

2.9.3

If "Other Airflow 2 version" selected, which one?

No response

What happened?

When setting global

AIRFLOW__OPERATORS__DEFAULT_DEFERRABLE =true
EmrCreateJobFlowOperator(wait_for_completion=false)

The operator should adds a deferral trigger to wait for Jobflow create completion status.

What you think should happen instead?

Since we did not ask to wait for completion, the deferrable argument should do nothing. The deferrable argument should only start a trigger when wait_for_completion=true

How to reproduce

See deferrable code here:

airflow/airflow/providers/amazon/aws/operators/emr.py

Lines 827 to 849 in 460d209

    
           if self.deferrable: 
        
               self.defer( 
        
                   trigger=EmrCreateJobFlowTrigger( 
        
                       job_flow_id=self._job_flow_id, 
        
                       aws_conn_id=self.aws_conn_id, 
        
                       waiter_delay=self.waiter_delay, 
        
                       waiter_max_attempts=self.waiter_max_attempts, 
        
                   ), 
        
                   method_name="execute_complete", 
        
                   # timeout is set to ensure that if a trigger dies, the timeout does not restart 
        
                   # 60 seconds is added to allow the trigger to exit gracefully (i.e. yield TriggerEvent) 
        
                   timeout=timedelta(seconds=self.waiter_max_attempts * self.waiter_delay + 60), 
        
               ) 
        
           if self.wait_for_completion: 
        
               self._emr_hook.get_waiter("job_flow_waiting").wait( 
        
                   ClusterId=self._job_flow_id, 
        
                   WaiterConfig=prune_dict( 
        
                       { 
        
                           "Delay": self.waiter_delay, 
        
                           "MaxAttempts": self.waiter_max_attempts, 
        
                       } 
        
                   ), 
        
               )

Sample code

EmrCreateJobFlowOperator(
        task_id="cluster_m5",
        wait_for_completion=false,
        deferrable=true
)

Operating System

Mac OS 14.5 (23F79)

Versions of Apache Airflow Providers

No response

Deployment

Astronomer

Deployment details

No response

Anything else?

No response

Are you willing to submit PR?

Yes I am willing to submit a PR!

Code of Conduct

I agree to follow this project's Code of Conduct

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

@eladkal

hello @eladkal , I am using WSL2 set up can I work on this issue here OS is Mac ? .If yes , I would like to work on this issue as my first contribution.

May I work on this issue with a little bit of your guidance?

@vincbeck

cc @vincbeck I think you fixed similar problem for emr serverless?

Correct. It seems many operators have the same issue. This is a very good first issue so please feel free to work on it and tag me to review it :) Here are some examples: #41191, #41103, #41059

@laksh-krishna-sharma

@laksh-krishna-sharma Please look at the example I provided, that should help you to fix this issue

@vincbeck

Thank you, @vincbeck . I will start working on it immediately. I will reach out if I have any questions.

@laksh-krishna-sharma

@laksh-krishna-sharma Do you want to finish your PR? I can give you some guidance with the remaining test cases that failed, if you want to

@dominikhei

sure sir @dominikhei

@dominikhei

Hi @dominikhei, may I work on this issue again? I’ve gained more experience since last time and feel confident I can handle it now

Navigation Menu

Repository navigation

EmrCreateJobFlowOperator should not wait_for_completion if wait_for_completion=false deferrable=true #40966

Apache Airflow version

If "Other Airflow 2 version" selected, which one?

What happened?

What you think should happen instead?

How to reproduce

Operating System

Versions of Apache Airflow Providers

Deployment

Deployment details

Anything else?

Are you willing to submit PR?

Code of Conduct

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

	if self.deferrable:
	self.defer(
	trigger=EmrCreateJobFlowTrigger(
	job_flow_id=self._job_flow_id,
	aws_conn_id=self.aws_conn_id,
	waiter_delay=self.waiter_delay,
	waiter_max_attempts=self.waiter_max_attempts,
	),
	method_name="execute_complete",
	# timeout is set to ensure that if a trigger dies, the timeout does not restart
	# 60 seconds is added to allow the trigger to exit gracefully (i.e. yield TriggerEvent)
	timeout=timedelta(seconds=self.waiter_max_attempts * self.waiter_delay + 60),
	)
	if self.wait_for_completion:
	self._emr_hook.get_waiter("job_flow_waiting").wait(
	ClusterId=self._job_flow_id,
	WaiterConfig=prune_dict(
	{
	"Delay": self.waiter_delay,
	"MaxAttempts": self.waiter_max_attempts,
	}
	),
	)

EmrCreateJobFlowOperator should not wait_for_completion if wait_for_completion=false deferrable=true #40966

Description

Apache Airflow version

If "Other Airflow 2 version" selected, which one?

What happened?

What you think should happen instead?

How to reproduce

Operating System

Versions of Apache Airflow Providers

Deployment

Deployment details

Anything else?

Are you willing to submit PR?

Code of Conduct

Activity

boring-cyborg commented on Jul 23, 2024

Prab-27 commented on Jul 26, 2024

laksh-krishna-sharma commented on Jul 30, 2024

eladkal commented on Aug 12, 2024

vincbeck commented on Aug 12, 2024

vincbeck commented on Aug 12, 2024

laksh-krishna-sharma commented on Aug 13, 2024

dominikhei commented on Jul 3, 2025

laksh-krishna-sharma commented on Jul 3, 2025

laksh-krishna-sharma commented on Sep 24, 2025

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions