You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "beobest2 (via GitHub)" <gi...@apache.org> on 2024/04/01 20:15:38 UTC
[PR] Fix automatic termination issue in EmrOperator by ensuring `waiter_max_attempts` is set for deferrable triggers [airflow]
beobest2 opened a new pull request, #38658:
URL: https://github.com/apache/airflow/pull/38658
When running EMR jobs that require a long duration, setting the `waiter_max_attempts` in EmrOperator to a high value does not prevent the "waiter error: max attempts" error from occurring after a certain period.
```
[2024-03-27 16:42:08,646] Trigger failed:
Traceback (most recent call last):
File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/jobs/triggerer_job_runner.py", line 529, in cleanup_finished_triggers
result = details["task"].result()
^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/jobs/triggerer_job_runner.py", line 601, in run_trigger
async for event in trigger.run():
File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/providers/amazon/aws/triggers/base.py", line 143, in run
await async_wait(
File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/providers/amazon/aws/utils/waiter_with_logging.py", line 133, in async_wait
raise AirflowException("Waiter error: max attempts reached")
```
---
An issue has been identified with the AWS provider code.
The `EmrOperator` that calls this trigger does not allow users to input `waiter_max_attempts`, resulting in the application of the default value of 600.
`airflow/providers/amazon/aws/triggers/emr.py`
```python
class EmrContainerTrigger(AwsBaseWaiterTrigger):
"""
Poll for the status of EMR container until reaches terminal state.
:param virtual_cluster_id: Reference Emr cluster id
:param job_id: job_id to check the state
:param aws_conn_id: Reference to AWS connection id
:param waiter_delay: polling period in seconds to check for the status
"""
def __init__(
self,
virtual_cluster_id: str,
job_id: str,
aws_conn_id: str | None = "aws_default",
poll_interval: int | None = None, # deprecated
waiter_delay: int = 30,
waiter_max_attempts: int = 600, # <- We can't modify this ..
):
```
In other words, if the `poll_interval` is set to 60 seconds, the job will terminate after 10 hours (1 minute * 600) regardless of the `waiter_max_attempts` value specified by the user.
To address this, modifications were made in `EmrOperator` to ensure that parameters provided by the user are also passed to the `EmrContainerTrigger`.
<!--
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements. See the NOTICE file
distributed with this work for additional information
regarding copyright ownership. The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
-->
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of an existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix automatic termination issue in EmrOperator by ensuring `waiter_max_attempts` is set for deferrable triggers [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on PR #38658:
URL: https://github.com/apache/airflow/pull/38658#issuecomment-2096508969
> @o-nikolas Previously, I was unable to properly test in my development environment because different dependency errors occurred. After solving the problem and running tests, I found that when deferrable is set to True, the `call_count` is not recorded as many times as the actual number of attempts we specified. Do you have an idea how can we verify the `call_count` in a unit test when using deferrable as it is now?
CC @syedahsn
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix automatic termination issue in EmrOperator by ensuring `waiter_max_attempts` is set for deferrable triggers [airflow]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #38658:
URL: https://github.com/apache/airflow/pull/38658#issuecomment-2123237486
Awesome work, congrats on your first merged pull request! You are invited to check our [Issue Tracker](https://github.com/apache/airflow/issues) for additional contributions.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix automatic termination issue in EmrOperator by ensuring `waiter_max_attempts` is set for deferrable triggers [airflow]
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #38658:
URL: https://github.com/apache/airflow/pull/38658#discussion_r1608684902
##########
airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -617,6 +617,14 @@ def execute(self, context: Context) -> str | None:
job_id=self.job_id,
aws_conn_id=self.aws_conn_id,
waiter_delay=self.poll_interval,
+ waiter_max_attempts=self.max_polling_attempts,
+ )
+ if self.max_polling_attempts
Review Comment:
Alright I guess making those changes would go beyond the scope of this PR, so I agree. Great work with the deep dive!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix automatic termination issue in EmrOperator by ensuring `waiter_max_attempts` is set for deferrable triggers [airflow]
Posted by "beobest2 (via GitHub)" <gi...@apache.org>.
beobest2 commented on PR #38658:
URL: https://github.com/apache/airflow/pull/38658#issuecomment-2053905730
@eladkal I have also added unit tests. Please review
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix automatic termination issue in EmrOperator by ensuring `waiter_max_attempts` is set for deferrable triggers [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck merged PR #38658:
URL: https://github.com/apache/airflow/pull/38658
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix automatic termination issue in EmrOperator by ensuring `waiter_max_attempts` is set for deferrable triggers [airflow]
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #38658:
URL: https://github.com/apache/airflow/pull/38658#discussion_r1603749424
##########
airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -617,6 +617,14 @@ def execute(self, context: Context) -> str | None:
job_id=self.job_id,
aws_conn_id=self.aws_conn_id,
waiter_delay=self.poll_interval,
+ waiter_max_attempts=self.max_polling_attempts,
+ )
+ if self.max_polling_attempts
Review Comment:
Hmm that's a good point. In that case, maybe you can edit the Trigger code to accept `None` (i.e. `waiter_max_attempts: int | None = 600`) as well, and do (I also included the `waiter_delay` param as well)
```
waiter_max_attempts = waiter_max_attempts or 600
waiter_delay = waiter_delay or 30
```
You would then be able to pass just pass `max_polling_attempts` as is, and the Trigger will handle the `None` case. What do you think about this approach?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix automatic termination issue in EmrOperator by ensuring `waiter_max_attempts` is set for deferrable triggers [airflow]
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #38658:
URL: https://github.com/apache/airflow/pull/38658#discussion_r1602266425
##########
airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -617,6 +617,14 @@ def execute(self, context: Context) -> str | None:
job_id=self.job_id,
aws_conn_id=self.aws_conn_id,
waiter_delay=self.poll_interval,
+ waiter_max_attempts=self.max_polling_attempts,
+ )
+ if self.max_polling_attempts
Review Comment:
Wouldn't this be a better solution?
```
waiter_max_attempts=self.max_polling_attempts if self.max_polling_attempts else 600
```
You can write a test where you provide a `max_polling_attempts`, and one where you don't, and assert that the correct value is used.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix automatic termination issue in EmrOperator by ensuring `waiter_max_attempts` is set for deferrable triggers [airflow]
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #38658:
URL: https://github.com/apache/airflow/pull/38658#discussion_r1602254073
##########
tests/providers/amazon/aws/sensors/test_emr_containers.py:
##########
@@ -84,3 +84,15 @@ def test_sensor_defer(self, mock_poke):
assert isinstance(
e.value.trigger, EmrContainerTrigger
), f"{e.value.trigger} is not a EmrContainerTrigger"
+
+ @mock.patch.object(
+ EmrContainerHook, "check_query_status", return_value=EmrContainerHook.INTERMEDIATE_STATES[0]
+ )
+ def test_sensor_defer_with_timeout(self, mock_check_query_status):
+ self.sensor.deferrable = True
+ self.sensor.max_polling_attempts = 1000
+
+ with pytest.raises(TaskDeferred):
Review Comment:
The problem with this test is that `check_query_status` is never called, because the sensor calls `self.defer` which raises the `TaskDeferred` Exception (which you are checking for), which ends the execution (i.e. the trigger code is never run).
What I would do here is just make sure that the Trigger is being called with the correct params (i.e. the supplied `max_polling_attempts`). The fact that the waiter calls the endpoint the correct number of times is technically the Trigger's responsibility. Because you are not overriding the `run` method of the `AwsBaseTrigger`, you should be fine with just ensuring that the correct parameters are being passed to the Trigger. The `run` method for the `AwsBaseTrigger` is tested in one [location](https://github.com/apache/airflow/blob/main/tests/providers/amazon/aws/triggers/test_base.py#L116).
The same is true for the operator test above.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix automatic termination issue in EmrOperator by ensuring `waiter_max_attempts` is set for deferrable triggers [airflow]
Posted by "beobest2 (via GitHub)" <gi...@apache.org>.
beobest2 commented on code in PR #38658:
URL: https://github.com/apache/airflow/pull/38658#discussion_r1602586826
##########
airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -617,6 +617,14 @@ def execute(self, context: Context) -> str | None:
job_id=self.job_id,
aws_conn_id=self.aws_conn_id,
waiter_delay=self.poll_interval,
+ waiter_max_attempts=self.max_polling_attempts,
+ )
+ if self.max_polling_attempts
Review Comment:
@syedahsn I felt that using hardcoded values (such as 600) in the code can be problematic for maintenance. Especially when such a value is referenced in multiple places, if a change is necessary, finding and modifying all the reference points increases the likelihood of making errors. How do you think about this point?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix automatic termination issue in EmrOperator by ensuring `waiter_max_attempts` is set for deferrable triggers [airflow]
Posted by "beobest2 (via GitHub)" <gi...@apache.org>.
beobest2 commented on PR #38658:
URL: https://github.com/apache/airflow/pull/38658#issuecomment-2092106048
@o-nikolas Previously, I was unable to properly test in my development environment because different dependency errors occurred. After solving the problem and running tests, I found that when deferrable is set to True, the `call_count` is not recorded as many times as the actual number of attempts we specified. Do you have an idea how can we verify the `call_count` in a unit test when using deferrable as it is now?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix automatic termination issue in EmrOperator by ensuring `waiter_max_attempts` is set for deferrable triggers [airflow]
Posted by "beobest2 (via GitHub)" <gi...@apache.org>.
beobest2 commented on code in PR #38658:
URL: https://github.com/apache/airflow/pull/38658#discussion_r1586861317
##########
tests/providers/amazon/aws/operators/test_emr_containers.py:
##########
@@ -144,6 +144,20 @@ def test_operator_defer(self, mock_submit_job, mock_check_query_status):
exc.value.trigger, EmrContainerTrigger
), f"{exc.value.trigger} is not a EmrContainerTrigger"
+ @mock.patch.object(EmrContainerHook, "submit_job")
+ @mock.patch.object(
+ EmrContainerHook, "check_query_status", return_value=EmrContainerHook.INTERMEDIATE_STATES[0]
+ )
+ def test_operator_defer_with_timeout(self, mock_submit_job, mock_check_query_status):
+ self.emr_container.deferrable = True
+ self.emr_container.max_polling_attempts = 1000
Review Comment:
@o-nikolas The purpose of this PR was to address the issue where setting this value high would not wait beyond the default of 600. By setting it to 1000, the modification is proven to be implemented. The `poll_interval` is set to 0 for testing, so it does not increase the time it takes to run the tests.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix automatic termination issue in EmrOperator by ensuring `waiter_max_attempts` is set for deferrable triggers [airflow]
Posted by "beobest2 (via GitHub)" <gi...@apache.org>.
beobest2 commented on code in PR #38658:
URL: https://github.com/apache/airflow/pull/38658#discussion_r1604169292
##########
airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -617,6 +617,14 @@ def execute(self, context: Context) -> str | None:
job_id=self.job_id,
aws_conn_id=self.aws_conn_id,
waiter_delay=self.poll_interval,
+ waiter_max_attempts=self.max_polling_attempts,
+ )
+ if self.max_polling_attempts
Review Comment:
@syedahsn
What you just mentioned was my initial approach.
However, after analyzing the code, I found that modifying only the input type of [EmrStepSensorTrigger](https://github.com/apache/airflow/blob/main/airflow/providers/amazon/aws/triggers/emr.py#L213) results in a discrepancy with its parent class, violating the Liskov Substitution Principle (LSP). If we were to modify the parent class, [AwsBaseWaiterTrigger](https://github.com/apache/airflow/blob/main/airflow/providers/amazon/aws/triggers/base.py#L31), we would also need to change the type of `waiter_max_attempts` in the [async_await](https://github.com/apache/airflow/blob/main/airflow/providers/amazon/aws/utils/waiter_with_logging.py#L88) utility that accepts this input. This would require handling the exception when `waiter_max_attempts` is None.
(https://github.com/apache/airflow/blob/main/airflow/providers/amazon/aws/utils/waiter_with_logging.py#L119)
Additionally, this function is widely used across AWS operators, and we cannot predict what other side effects it might cause.
Considering these factors, I believe the current code modification, while not entirely clean, is the best approach.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix automatic termination issue in EmrOperator by ensuring `waiter_max_attempts` is set for deferrable triggers [airflow]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #38658:
URL: https://github.com/apache/airflow/pull/38658#issuecomment-2030484929
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]( https://github.com/apache/airflow/blob/main/contributing-docs/08_static_code_checks.rst#prerequisites-for-pre-commit-hooks) 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](https://github.com/apache/airflow/blob/main/docs/apache-airflow/howto/custom-operator.rst) Consider adding an example DAG that shows how users should use it.
- Consider using [Breeze environment](https://github.com/apache/airflow/blob/main/dev/breeze/doc/README.rst) 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](https://www.apache.org/foundation/policies/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]( https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#coding-style-and-best-practices).
- 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: dev@airflow.apache.org
Slack: https://s.apache.org/airflow-slack
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix automatic termination issue in EmrOperator by ensuring `waiter_max_attempts` is set for deferrable triggers [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on PR #38658:
URL: https://github.com/apache/airflow/pull/38658#issuecomment-2044967137
can you add unit tests to cover this change?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix automatic termination issue in EmrOperator by ensuring `waiter_max_attempts` is set for deferrable triggers [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on PR #38658:
URL: https://github.com/apache/airflow/pull/38658#issuecomment-2091768196
Running workflows again after recent commits
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix automatic termination issue in EmrOperator by ensuring `waiter_max_attempts` is set for deferrable triggers [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on code in PR #38658:
URL: https://github.com/apache/airflow/pull/38658#discussion_r1586906672
##########
tests/providers/amazon/aws/operators/test_emr_containers.py:
##########
@@ -144,6 +144,20 @@ def test_operator_defer(self, mock_submit_job, mock_check_query_status):
exc.value.trigger, EmrContainerTrigger
), f"{exc.value.trigger} is not a EmrContainerTrigger"
+ @mock.patch.object(EmrContainerHook, "submit_job")
+ @mock.patch.object(
+ EmrContainerHook, "check_query_status", return_value=EmrContainerHook.INTERMEDIATE_STATES[0]
+ )
+ def test_operator_defer_with_timeout(self, mock_submit_job, mock_check_query_status):
+ self.emr_container.deferrable = True
+ self.emr_container.max_polling_attempts = 1000
Review Comment:
Ahh, the poll_interval is set to 0 in the setup_method, missed that. Looks good :+1:
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix automatic termination issue in EmrOperator by ensuring `waiter_max_attempts` is set for deferrable triggers [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on PR #38658:
URL: https://github.com/apache/airflow/pull/38658#issuecomment-2089264782
Tests need to pass after your rebase from main. Otherwise looks good to merge
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix automatic termination issue in EmrOperator by ensuring `waiter_max_attempts` is set for deferrable triggers [airflow]
Posted by "beobest2 (via GitHub)" <gi...@apache.org>.
beobest2 commented on code in PR #38658:
URL: https://github.com/apache/airflow/pull/38658#discussion_r1602632380
##########
tests/providers/amazon/aws/sensors/test_emr_containers.py:
##########
@@ -84,3 +84,15 @@ def test_sensor_defer(self, mock_poke):
assert isinstance(
e.value.trigger, EmrContainerTrigger
), f"{e.value.trigger} is not a EmrContainerTrigger"
+
+ @mock.patch.object(
+ EmrContainerHook, "check_query_status", return_value=EmrContainerHook.INTERMEDIATE_STATES[0]
+ )
+ def test_sensor_defer_with_timeout(self, mock_check_query_status):
+ self.sensor.deferrable = True
+ self.sensor.max_polling_attempts = 1000
+
+ with pytest.raises(TaskDeferred):
Review Comment:
I have modified the test code to only verify that the correct parameters are being passed to the trigger.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix automatic termination issue in EmrOperator by ensuring `waiter_max_attempts` is set for deferrable triggers [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on code in PR #38658:
URL: https://github.com/apache/airflow/pull/38658#discussion_r1585509531
##########
tests/providers/amazon/aws/operators/test_emr_containers.py:
##########
@@ -144,6 +144,20 @@ def test_operator_defer(self, mock_submit_job, mock_check_query_status):
exc.value.trigger, EmrContainerTrigger
), f"{exc.value.trigger} is not a EmrContainerTrigger"
+ @mock.patch.object(EmrContainerHook, "submit_job")
+ @mock.patch.object(
+ EmrContainerHook, "check_query_status", return_value=EmrContainerHook.INTERMEDIATE_STATES[0]
+ )
+ def test_operator_defer_with_timeout(self, mock_submit_job, mock_check_query_status):
+ self.emr_container.deferrable = True
+ self.emr_container.max_polling_attempts = 1000
Review Comment:
1000? How long does this unittest take? That seems excessive.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix automatic termination issue in EmrOperator by ensuring `waiter_max_attempts` is set for deferrable triggers [airflow]
Posted by "beobest2 (via GitHub)" <gi...@apache.org>.
beobest2 commented on code in PR #38658:
URL: https://github.com/apache/airflow/pull/38658#discussion_r1602586826
##########
airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -617,6 +617,14 @@ def execute(self, context: Context) -> str | None:
job_id=self.job_id,
aws_conn_id=self.aws_conn_id,
waiter_delay=self.poll_interval,
+ waiter_max_attempts=self.max_polling_attempts,
+ )
+ if self.max_polling_attempts
Review Comment:
@syedahsn I felt that using hardcoded values (such as 600) in the code can be problematic for maintenance. Especially when such a value is referenced in multiple places, if a change is necessary, finding and modifying all the reference points increases the likelihood of making errors. Therefore, I chose to avoid defining defaults directly in the code.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix automatic termination issue in EmrOperator by ensuring `waiter_max_attempts` is set for deferrable triggers [airflow]
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #38658:
URL: https://github.com/apache/airflow/pull/38658#discussion_r1603754080
##########
tests/providers/amazon/aws/sensors/test_emr_containers.py:
##########
@@ -84,3 +84,17 @@ def test_sensor_defer(self, mock_poke):
assert isinstance(
e.value.trigger, EmrContainerTrigger
), f"{e.value.trigger} is not a EmrContainerTrigger"
+
+ @mock.patch("airflow.providers.amazon.aws.sensors.emr.EmrContainerSensor.poke")
+ def test_sensor_defer_with_timeout(self, mock_poke):
+ self.sensor.deferrable = True
+ mock_poke.return_value = False
+ self.sensor.max_retries = 1000
+
+ with pytest.raises(TaskDeferred) as e:
+ self.sensor.execute(context=None)
+
+ trigger = e.value.trigger
+ assert isinstance(trigger, EmrContainerTrigger), f"{trigger} is not a EmrContainerTrigger"
+ assert trigger.waiter_delay == self.sensor.poll_interval
Review Comment:
Nice! This looks great.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix automatic termination issue in EmrOperator by ensuring `waiter_max_attempts` is set for deferrable triggers [airflow]
Posted by "beobest2 (via GitHub)" <gi...@apache.org>.
beobest2 commented on code in PR #38658:
URL: https://github.com/apache/airflow/pull/38658#discussion_r1604169292
##########
airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -617,6 +617,14 @@ def execute(self, context: Context) -> str | None:
job_id=self.job_id,
aws_conn_id=self.aws_conn_id,
waiter_delay=self.poll_interval,
+ waiter_max_attempts=self.max_polling_attempts,
+ )
+ if self.max_polling_attempts
Review Comment:
@syedahsn
What you just mentioned was my initial approach.
However, after analyzing the code, I found that modifying only the input type of `[EmrStepSensorTrigger](https://github.com/apache/airflow/blob/main/airflow/providers/amazon/aws/triggers/emr.py#L213)` results in a discrepancy with its parent class, violating the Liskov Substitution Principle (LSP). If we were to modify the parent class, `[AwsBaseWaiterTrigger](https://github.com/apache/airflow/blob/main/airflow/providers/amazon/aws/triggers/base.py#L31)`, we would also need to change the type of `waiter_max_attempts` in the `[async_await](https://github.com/apache/airflow/blob/main/airflow/providers/amazon/aws/utils/waiter_with_logging.py#L88)` utility that accepts this input. This would require handling the exception when `waiter_max_attempts` is None.
(https://github.com/apache/airflow/blob/main/airflow/providers/amazon/aws/utils/waiter_with_logging.py#L119)
Additionally, this function is widely used across AWS operators, and we cannot predict what other side effects it might cause.
Considering these factors, I believe the current code modification, while not entirely clean, is the best approach.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org