Fix EmrCreateJobFlowOperator using deferrable mode with wait_for_completion

[laksh-krishna-sharma]

---

Summary of Changes

This PR addresses issue #40966 by modifying the logic in the EmrCreateJobFlowOperator to ensure that the deferrable argument only activates the deferral trigger when wait_for_completion is also set to True. Previously, the deferral trigger would be added regardless of the wait_for_completion setting, leading to unintended behavior.

Changes Made:

• Updated the condition for deferring to check both deferrable and wait_for_completion.
• Adjusted the condition for synchronous waiting to occur only when wait_for_completion=True and deferrable=False.

Please review the changes, and let me know if further adjustments are needed.

Closes: #40966

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide (https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
Here are some useful points:

• Pay attention to the quality of your code (ruff, mypy and type annotations). Our pre-commits will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
• Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
• Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: [email protected]
  Slack: https://s.apache.org/airflow-slack

[shahar1]

Welcome to Apache Airflow and congrats for your first PR!
Good start, I have some comments :)

[shahar1]

1. I'd consider refactoring both conditions as follows (a bit more readable IMO):

if self.wait_for_completion:
    if self.deferrable:
       ...
    else:
       ...

2. Unit tests + docstring update for this behavior would be helpful.

[vincbeck]

Is not it the case for 1? But I definitely agree on 2 :)

[laksh-krishna-sharma]

Thank you for letting me know, @shahar1. I will make the changes and add comments again . I'll update you once the problem is resolved."

[laksh-krishna-sharma]

@shahar1 sir please review the update.

[vincbeck]

Can you please add unit tests?

[laksh-krishna-sharma]

@shahar1 sir could you please advise on the appropriate folder for placing unit test cases?

[shahar1]

@shahar1 sir could you please advise on the appropriate folder for placing unit test cases?

tests/providers/amazon/aws/operators/, and there check all the files that start with emr_.

[laksh-krishna-sharma]

@shahar1 sir and @vincbeck sir I've added unit test cases and done my best to cover the expected behaviors. As I'm a beginner, I'd greatly appreciate your feedback and any suggestions for improvement.

[vincbeck]

Good job!

[vincbeck]

@shahar1 sir and @vincbeck sir I've added unit test cases and done my best to cover the expected behaviors. As I'm a beginner, I'd greatly appreciate your feedback and any suggestions for improvement.

Thanks for doing it! The tests are pretty good!

[shahar1]

Looks great!

[vincbeck]

Some tests are failing, could you check that please?

[vincbeck]

https://github.com/apache/airflow/actions/runs/10561464118/job/29259176171?pr=41561

[laksh-krishna-sharma]

Thank you for your review @vincbeck and @shahar1. I will investigate the issue and address the failing tests as soon as possible. I'll update once the problem is resolved.

[laksh-krishna-sharma]

@shahar1 sir and @vincbeck sir I've added unit test cases and done my best to cover the expected behaviors. As I'm a beginner, I'd greatly appreciate your feedback and any suggestions for improvement but i am stuck in resolving these errors please sir can you guide me to resolve these issue

[shahar1]

@shahar1 sir and @vincbeck sir I've added unit test cases and done my best to cover the expected behaviors. As I'm a beginner, I'd greatly appreciate your feedback and any suggestions for improvement but i am stuck in resolving these errors please sir can you guide me to resolve these issue

Hello, here's some guidance:

1. In test_create_job_flow_deferrable you need to set self.operator.wait_for_completion = True (otherwise it won't pass the first if).
2. In test_deferrable_and_wait_for_completion, when you call the assert statement you create a new instance of EmrCreateJobFlowTrigger - which will always be different than the instance that is created by the original function.
   I'd suggest to mock EmrCreateJobFlowTrigger (let's call it mock_trigger) and then you could assert on trigger=mock_trigger.return_value.
   You could also create a separate assertion for the attributes of mock_trigger to ensure that all the parameters are as expected.

Also, please notice that you need to resolve conflicts by merging/rebasing from main.
Please let me know if you encounter any issues.

[laksh-krishna-sharma]

Thank you for your help, @shahar1 sir. I’ve resolved the issue, and all checks have now passed on GitHub. I really appreciate your guidance!

[vincbeck]

There is a conflict now :)

[laksh-krishna-sharma]

Sir, I resolved the branch conflict. Please review the updates.

[laksh-krishna-sharma]

@shahar1 sir and @vincbeck sir I've tried to resolve conflict and done my best to cover the expected behaviors. As I'm a beginner, I'd greatly appreciate your feedback and any suggestions for improvement but i am stuck in resolving these errors please sir can you guide me to resolve these issue