Increase waiter_max_attempts default value in EcsRunTaskOperator

[mjsqu]

---

closes: #33711
related: #33016, #30586

Prior to v8.0.0 the default waiter_max_attempts parameter in EcsRunTaskOperator was set to sys.maxsize. The value is a large integer, which should always exceed the execution_timeout value for a given task. I believe this is why the line setting the value to sys.maxsize is accompanied by a comment:

waiter.config.max_attempts=sys.maxsize# timeout is managed by airflow

Changes in v8.0.0 introduced a configurable value defaulting to 100 for this parameter, with a delay defaulting to 6 seconds (the changes were introduced to resolve#30586). The line above remains, but I think it is being overridden by the call to waiter.wait:

waiter.wait(
cluster=self.cluster,
tasks=[self.arn],
WaiterConfig={
"Delay": self.waiter_delay,
"MaxAttempts": self.waiter_max_attempts, # !!! Overriding previously set max_attempts?
},
)

The defaults for these previously unavailable parameters have been set to:

waiter_delay: int=6,
waiter_max_attempts: int=100,

I believe the introduction of a default value of 100 has changed the behaviour of the operator - now when not specified, all tasks detach from the running ECS Task at 10 minutes, reporting failed status back to Airflow.

The fix replaces the default value with sys.maxsize - which should revert the waiter functionality back to prior working versions.

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

[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 Contribution Guide (https://github.com/apache/airflow/blob/main/CONTRIBUTING.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: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[mjsqu]

While I wait for my sleep 800 command to run having manually set waiter_max_attempts to sys.maxsize...

Here's the documentation for the ECS TasksStopped waiter: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ecs/waiter/TasksStopped.html

If not set, it has defaults of

WaiterConfig={
'Delay': 6,
'MaxAttempts': 100
}

too. I think this was the motivation for setting these as the defaults, however the prior functionality would override the MaxAttempts to a really high number and let Airflow's execution_timeout parameter handle long-running tasks. The execution_timeout method works well, when the Airflow timeout is hit, the ECS task is killed.

[hussein-awala]

There are two things here:

1. Changing the default value to infinite could be a breaking change (cc: @eladkal )
2. If it's okay to change it, the best way is to make it None and replace the loop in the Trigger with while True when it's None.

[potiuk]

Agree it might be seen as breaking but looking at the problem description I think the previous change was accidentally breaking (i.e. changing from effectively infinite to finite) and this one is simply a bugfix restoring it.

[mjsqu]

Thanks for the review. I agree that adding sys.maxsize must have been a workaround and is a bit non-standard so perhaps this PR might be able to resolve that by using some other method.

I had considered if the max_attempts and delay parameters could be calculated at runtime by taking into consideration the execution_timeout value, however this would require a bit more of a rewrite of the Operator.

[eladkal]

Agree it might be seen as breaking but looking at the problem description I think the previous change was accidentally breaking (i.e. changing from effectively infinite to finite) and this one is simply a bugfix restoring it.

Then I can accept this as non breaking change but a bug fix. Yet requesting to add a note explaining it in the top of the CHANGELOG. I will bake it to the proper release version during release process.

[mjsqu]

Changelog note added 456ee50

Unsure of the best way to refer back to the original PR that I am partially reverting, I trust you'll properly format the Changelog message when it comes to release time. TIA.

[eladkal]

Lets modify this to a more explanatory text.
Dont mention PR numbers. The commit title that we add during release cycle will do that and refer to this PR for more details.

What we are seeking in the note is just a simple explnatipn of the situation "A bug intoduced in provider version... Caused.... In this version we are fixing it by returning value to ..."
Something like this.

[eladkal]

LGTM

needs rebase and resolve conflicts

[mjsqu]

LGTM

needs rebase and resolve conflicts

Resolved conflict in CHANGELOG.rst - 56782d5

[eladkal]

Tests fail :(

[mjsqu]

Tests fail :(

Tests were failing due to the size of sys.maxsize. The waiter_max_attempts value is used in a timedelta calculation here (https://github.com/mjsqu/airflow/blob/b9453e9674e69203ca8658aa15f7e5c24e88f4db/airflow/providers/amazon/aws/operators/ecs.py#L564)

>>>timedelta(seconds=sys.maxsize*delay+60)
Traceback (mostrecentcalllast):
File"<stdin>", line1, in<module>OverflowError: PythoninttoolargetoconverttoCint

To resolve this, I have set the default waiter_max_attempts value high enough such that the waiter would run for 1M years. This allows the timedelta calculation to succeed, while providing a value that is likely longer than any ECS Tasks will ever run for.

max_attempts=1000000*365*24*60*10>>>timedelta(seconds=max_attempts*delay+60)
datetime.timedelta(days=365000000, seconds=60)

Alternatively, if possible, it would be good to calculate a default_max_attempts at runtime using execution_timeout, I'm just unsure how.

I'm open to suggestions about other values to use instead of 1 million years, there are lower values that would be equally sensible.

[mjsqu]

Latest test failures:

cluster="c", tasks=["arn"], WaiterConfig={"Delay": 6, "MaxAttempts": 1000000*365*24*60*10}
)
>assert1000000*365*24*60*10==client_mock.get_waiter.return_value.config.max_attemptsEAssertionError: assert ((((1000000*365) *24) *60) *10) ==<MagicMockname='client.get_waiter().config.max_attempts'id='139852189801392'>E+where<MagicMockname='client.get_waiter().config.max_attempts'id='139852189801392'>=<MagicMockname='client.get_waiter().config'id='139852190309584'>.max_attemptsE+where<MagicMockname='client.get_waiter().config'id='139852190309584'>=<MagicMockname='client.get_waiter()'id='139852190269200'>.configE+where<MagicMockname='client.get_waiter()'id='139852190269200'>=<MagicMockname='client.get_waiter'id='139852190252720'>.return_valueE+where<MagicMockname='client.get_waiter'id='139852190252720'>=<MagicMockname='client'id='139852190228688'>.get_waiter

Struggling to understand how the assertion fails, perhaps there's a precision issue such that client_mock.get_waiter.return_value.config.max_attempts does not return the exact value it was provided with?

[mjsqu]

Anyone following this PR/Issue, we have set the waiter_max_attempts value to sys.maxsize in the interim for affected EcsRunTaskOperator tasks, but you could choose a sufficiently large value to force the waiter to run for longer than execution_timeout:

waiter_max_attempts * waiter_delay > execution_timeout

e.g. for an execution_timeout of 1 hour (3600 seconds), set:

waiter_max_attempts=650, # Greater than 600waiter_delay=6,
# Total waiter time = 3,900 seconds == 65 minutes

[hussein-awala]

The failed check was not valid after removing this line.

[vincbeck]

Thanks for fixing that bug!

[boring-cyborg]

Awesome work, congrats on your first merged pull request! You are invited to check our Issue Tracker for additional contributions.

[mjsqu]

---

closes: #33711
related: #33016, #30586

Prior to v8.0.0 the default waiter_max_attempts parameter in EcsRunTaskOperator was set to sys.maxsize. The value is a large integer, which should always exceed the execution_timeout value for a given task. I believe this is why the line setting the value to sys.maxsize is accompanied by a comment:

waiter.config.max_attempts=sys.maxsize# timeout is managed by airflow

Changes in v8.0.0 introduced a configurable value defaulting to 100 for this parameter, with a delay defaulting to 6 seconds (the changes were introduced to resolve#30586). The line above remains, but I think it is being overridden by the call to waiter.wait:

waiter.wait(
cluster=self.cluster,
tasks=[self.arn],
WaiterConfig={
"Delay": self.waiter_delay,
"MaxAttempts": self.waiter_max_attempts, # !!! Overriding previously set max_attempts?
},
)

The defaults for these previously unavailable parameters have been set to:

waiter_delay: int=6,
waiter_max_attempts: int=100,

I believe the introduction of a default value of 100 has changed the behaviour of the operator - now when not specified, all tasks detach from the running ECS Task at 10 minutes, reporting failed status back to Airflow.

The fix replaces the default value with sys.maxsize - which should revert the waiter functionality back to prior working versions.

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

[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 Contribution Guide (https://github.com/apache/airflow/blob/main/CONTRIBUTING.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: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[mjsqu]

While I wait for my sleep 800 command to run having manually set waiter_max_attempts to sys.maxsize...

Here's the documentation for the ECS TasksStopped waiter: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ecs/waiter/TasksStopped.html

If not set, it has defaults of

WaiterConfig={
'Delay': 6,
'MaxAttempts': 100
}

too. I think this was the motivation for setting these as the defaults, however the prior functionality would override the MaxAttempts to a really high number and let Airflow's execution_timeout parameter handle long-running tasks. The execution_timeout method works well, when the Airflow timeout is hit, the ECS task is killed.

[hussein-awala]

There are two things here:

1. Changing the default value to infinite could be a breaking change (cc: @eladkal )
2. If it's okay to change it, the best way is to make it None and replace the loop in the Trigger with while True when it's None.

[potiuk]

Agree it might be seen as breaking but looking at the problem description I think the previous change was accidentally breaking (i.e. changing from effectively infinite to finite) and this one is simply a bugfix restoring it.

[mjsqu]

Thanks for the review. I agree that adding sys.maxsize must have been a workaround and is a bit non-standard so perhaps this PR might be able to resolve that by using some other method.

I had considered if the max_attempts and delay parameters could be calculated at runtime by taking into consideration the execution_timeout value, however this would require a bit more of a rewrite of the Operator.

[eladkal]

Agree it might be seen as breaking but looking at the problem description I think the previous change was accidentally breaking (i.e. changing from effectively infinite to finite) and this one is simply a bugfix restoring it.

Then I can accept this as non breaking change but a bug fix. Yet requesting to add a note explaining it in the top of the CHANGELOG. I will bake it to the proper release version during release process.

[mjsqu]

Changelog note added 456ee50

Unsure of the best way to refer back to the original PR that I am partially reverting, I trust you'll properly format the Changelog message when it comes to release time. TIA.

[eladkal]

Lets modify this to a more explanatory text.
Dont mention PR numbers. The commit title that we add during release cycle will do that and refer to this PR for more details.

What we are seeking in the note is just a simple explnatipn of the situation "A bug intoduced in provider version... Caused.... In this version we are fixing it by returning value to ..."
Something like this.

[eladkal]

LGTM

needs rebase and resolve conflicts

[mjsqu]

LGTM

needs rebase and resolve conflicts

Resolved conflict in CHANGELOG.rst - 56782d5

[eladkal]

Tests fail :(

[mjsqu]

Tests fail :(

Tests were failing due to the size of sys.maxsize. The waiter_max_attempts value is used in a timedelta calculation here (https://github.com/mjsqu/airflow/blob/b9453e9674e69203ca8658aa15f7e5c24e88f4db/airflow/providers/amazon/aws/operators/ecs.py#L564)

>>>timedelta(seconds=sys.maxsize*delay+60)
Traceback (mostrecentcalllast):
File"<stdin>", line1, in<module>OverflowError: PythoninttoolargetoconverttoCint

To resolve this, I have set the default waiter_max_attempts value high enough such that the waiter would run for 1M years. This allows the timedelta calculation to succeed, while providing a value that is likely longer than any ECS Tasks will ever run for.

max_attempts=1000000*365*24*60*10>>>timedelta(seconds=max_attempts*delay+60)
datetime.timedelta(days=365000000, seconds=60)

Alternatively, if possible, it would be good to calculate a default_max_attempts at runtime using execution_timeout, I'm just unsure how.

I'm open to suggestions about other values to use instead of 1 million years, there are lower values that would be equally sensible.

[mjsqu]

Latest test failures:

cluster="c", tasks=["arn"], WaiterConfig={"Delay": 6, "MaxAttempts": 1000000*365*24*60*10}
)
>assert1000000*365*24*60*10==client_mock.get_waiter.return_value.config.max_attemptsEAssertionError: assert ((((1000000*365) *24) *60) *10) ==<MagicMockname='client.get_waiter().config.max_attempts'id='139852189801392'>E+where<MagicMockname='client.get_waiter().config.max_attempts'id='139852189801392'>=<MagicMockname='client.get_waiter().config'id='139852190309584'>.max_attemptsE+where<MagicMockname='client.get_waiter().config'id='139852190309584'>=<MagicMockname='client.get_waiter()'id='139852190269200'>.configE+where<MagicMockname='client.get_waiter()'id='139852190269200'>=<MagicMockname='client.get_waiter'id='139852190252720'>.return_valueE+where<MagicMockname='client.get_waiter'id='139852190252720'>=<MagicMockname='client'id='139852190228688'>.get_waiter

Struggling to understand how the assertion fails, perhaps there's a precision issue such that client_mock.get_waiter.return_value.config.max_attempts does not return the exact value it was provided with?

[mjsqu]

Anyone following this PR/Issue, we have set the waiter_max_attempts value to sys.maxsize in the interim for affected EcsRunTaskOperator tasks, but you could choose a sufficiently large value to force the waiter to run for longer than execution_timeout:

waiter_max_attempts * waiter_delay > execution_timeout

e.g. for an execution_timeout of 1 hour (3600 seconds), set:

waiter_max_attempts=650, # Greater than 600waiter_delay=6,
# Total waiter time = 3,900 seconds == 65 minutes

[hussein-awala]

The failed check was not valid after removing this line.

[vincbeck]

Thanks for fixing that bug!

[boring-cyborg]

Awesome work, congrats on your first merged pull request! You are invited to check our Issue Tracker for additional contributions.

[mjsqu]

---

closes: #33711
related: #33016, #30586

Prior to v8.0.0 the default waiter_max_attempts parameter in EcsRunTaskOperator was set to sys.maxsize. The value is a large integer, which should always exceed the execution_timeout value for a given task. I believe this is why the line setting the value to sys.maxsize is accompanied by a comment:

waiter.config.max_attempts=sys.maxsize# timeout is managed by airflow

Changes in v8.0.0 introduced a configurable value defaulting to 100 for this parameter, with a delay defaulting to 6 seconds (the changes were introduced to resolve#30586). The line above remains, but I think it is being overridden by the call to waiter.wait:

waiter.wait(
cluster=self.cluster,
tasks=[self.arn],
WaiterConfig={
"Delay": self.waiter_delay,
"MaxAttempts": self.waiter_max_attempts, # !!! Overriding previously set max_attempts?
},
)

The defaults for these previously unavailable parameters have been set to:

waiter_delay: int=6,
waiter_max_attempts: int=100,

I believe the introduction of a default value of 100 has changed the behaviour of the operator - now when not specified, all tasks detach from the running ECS Task at 10 minutes, reporting failed status back to Airflow.

The fix replaces the default value with sys.maxsize - which should revert the waiter functionality back to prior working versions.

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

[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 Contribution Guide (https://github.com/apache/airflow/blob/main/CONTRIBUTING.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: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[mjsqu]

While I wait for my sleep 800 command to run having manually set waiter_max_attempts to sys.maxsize...

Here's the documentation for the ECS TasksStopped waiter: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ecs/waiter/TasksStopped.html

If not set, it has defaults of

WaiterConfig={
'Delay': 6,
'MaxAttempts': 100
}

too. I think this was the motivation for setting these as the defaults, however the prior functionality would override the MaxAttempts to a really high number and let Airflow's execution_timeout parameter handle long-running tasks. The execution_timeout method works well, when the Airflow timeout is hit, the ECS task is killed.

[hussein-awala]

There are two things here:

1. Changing the default value to infinite could be a breaking change (cc: @eladkal )
2. If it's okay to change it, the best way is to make it None and replace the loop in the Trigger with while True when it's None.

[potiuk]

Agree it might be seen as breaking but looking at the problem description I think the previous change was accidentally breaking (i.e. changing from effectively infinite to finite) and this one is simply a bugfix restoring it.

[mjsqu]

Thanks for the review. I agree that adding sys.maxsize must have been a workaround and is a bit non-standard so perhaps this PR might be able to resolve that by using some other method.

I had considered if the max_attempts and delay parameters could be calculated at runtime by taking into consideration the execution_timeout value, however this would require a bit more of a rewrite of the Operator.

[eladkal]

Agree it might be seen as breaking but looking at the problem description I think the previous change was accidentally breaking (i.e. changing from effectively infinite to finite) and this one is simply a bugfix restoring it.

Then I can accept this as non breaking change but a bug fix. Yet requesting to add a note explaining it in the top of the CHANGELOG. I will bake it to the proper release version during release process.

[mjsqu]

Changelog note added 456ee50

Unsure of the best way to refer back to the original PR that I am partially reverting, I trust you'll properly format the Changelog message when it comes to release time. TIA.

[eladkal]

Lets modify this to a more explanatory text.
Dont mention PR numbers. The commit title that we add during release cycle will do that and refer to this PR for more details.

What we are seeking in the note is just a simple explnatipn of the situation "A bug intoduced in provider version... Caused.... In this version we are fixing it by returning value to ..."
Something like this.

[eladkal]

LGTM

needs rebase and resolve conflicts

[mjsqu]

LGTM

needs rebase and resolve conflicts

Resolved conflict in CHANGELOG.rst - 56782d5

[eladkal]

Tests fail :(

[mjsqu]

Tests fail :(

Tests were failing due to the size of sys.maxsize. The waiter_max_attempts value is used in a timedelta calculation here (https://github.com/mjsqu/airflow/blob/b9453e9674e69203ca8658aa15f7e5c24e88f4db/airflow/providers/amazon/aws/operators/ecs.py#L564)

>>>timedelta(seconds=sys.maxsize*delay+60)
Traceback (mostrecentcalllast):
File"<stdin>", line1, in<module>OverflowError: PythoninttoolargetoconverttoCint

To resolve this, I have set the default waiter_max_attempts value high enough such that the waiter would run for 1M years. This allows the timedelta calculation to succeed, while providing a value that is likely longer than any ECS Tasks will ever run for.

max_attempts=1000000*365*24*60*10>>>timedelta(seconds=max_attempts*delay+60)
datetime.timedelta(days=365000000, seconds=60)

Alternatively, if possible, it would be good to calculate a default_max_attempts at runtime using execution_timeout, I'm just unsure how.

I'm open to suggestions about other values to use instead of 1 million years, there are lower values that would be equally sensible.

[mjsqu]

Latest test failures:

cluster="c", tasks=["arn"], WaiterConfig={"Delay": 6, "MaxAttempts": 1000000*365*24*60*10}
)
>assert1000000*365*24*60*10==client_mock.get_waiter.return_value.config.max_attemptsEAssertionError: assert ((((1000000*365) *24) *60) *10) ==<MagicMockname='client.get_waiter().config.max_attempts'id='139852189801392'>E+where<MagicMockname='client.get_waiter().config.max_attempts'id='139852189801392'>=<MagicMockname='client.get_waiter().config'id='139852190309584'>.max_attemptsE+where<MagicMockname='client.get_waiter().config'id='139852190309584'>=<MagicMockname='client.get_waiter()'id='139852190269200'>.configE+where<MagicMockname='client.get_waiter()'id='139852190269200'>=<MagicMockname='client.get_waiter'id='139852190252720'>.return_valueE+where<MagicMockname='client.get_waiter'id='139852190252720'>=<MagicMockname='client'id='139852190228688'>.get_waiter

Struggling to understand how the assertion fails, perhaps there's a precision issue such that client_mock.get_waiter.return_value.config.max_attempts does not return the exact value it was provided with?

[mjsqu]

Anyone following this PR/Issue, we have set the waiter_max_attempts value to sys.maxsize in the interim for affected EcsRunTaskOperator tasks, but you could choose a sufficiently large value to force the waiter to run for longer than execution_timeout:

waiter_max_attempts * waiter_delay > execution_timeout

e.g. for an execution_timeout of 1 hour (3600 seconds), set:

waiter_max_attempts=650, # Greater than 600waiter_delay=6,
# Total waiter time = 3,900 seconds == 65 minutes

[hussein-awala]

The failed check was not valid after removing this line.

[vincbeck]

Thanks for fixing that bug!

[boring-cyborg]

Awesome work, congrats on your first merged pull request! You are invited to check our Issue Tracker for additional contributions.

[mjsqu]

---

closes: #33711
related: #33016, #30586

Prior to v8.0.0 the default waiter_max_attempts parameter in EcsRunTaskOperator was set to sys.maxsize. The value is a large integer, which should always exceed the execution_timeout value for a given task. I believe this is why the line setting the value to sys.maxsize is accompanied by a comment:

waiter.config.max_attempts=sys.maxsize# timeout is managed by airflow

Changes in v8.0.0 introduced a configurable value defaulting to 100 for this parameter, with a delay defaulting to 6 seconds (the changes were introduced to resolve#30586). The line above remains, but I think it is being overridden by the call to waiter.wait:

waiter.wait(
cluster=self.cluster,
tasks=[self.arn],
WaiterConfig={
"Delay": self.waiter_delay,
"MaxAttempts": self.waiter_max_attempts, # !!! Overriding previously set max_attempts?
},
)

The defaults for these previously unavailable parameters have been set to:

waiter_delay: int=6,
waiter_max_attempts: int=100,

I believe the introduction of a default value of 100 has changed the behaviour of the operator - now when not specified, all tasks detach from the running ECS Task at 10 minutes, reporting failed status back to Airflow.

The fix replaces the default value with sys.maxsize - which should revert the waiter functionality back to prior working versions.

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

[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 Contribution Guide (https://github.com/apache/airflow/blob/main/CONTRIBUTING.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: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[mjsqu]

While I wait for my sleep 800 command to run having manually set waiter_max_attempts to sys.maxsize...

Here's the documentation for the ECS TasksStopped waiter: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ecs/waiter/TasksStopped.html

If not set, it has defaults of

WaiterConfig={
'Delay': 6,
'MaxAttempts': 100
}

too. I think this was the motivation for setting these as the defaults, however the prior functionality would override the MaxAttempts to a really high number and let Airflow's execution_timeout parameter handle long-running tasks. The execution_timeout method works well, when the Airflow timeout is hit, the ECS task is killed.

[hussein-awala]

There are two things here:

1. Changing the default value to infinite could be a breaking change (cc: @eladkal )
2. If it's okay to change it, the best way is to make it None and replace the loop in the Trigger with while True when it's None.

[potiuk]

Agree it might be seen as breaking but looking at the problem description I think the previous change was accidentally breaking (i.e. changing from effectively infinite to finite) and this one is simply a bugfix restoring it.

[mjsqu]

Thanks for the review. I agree that adding sys.maxsize must have been a workaround and is a bit non-standard so perhaps this PR might be able to resolve that by using some other method.

I had considered if the max_attempts and delay parameters could be calculated at runtime by taking into consideration the execution_timeout value, however this would require a bit more of a rewrite of the Operator.

[eladkal]

Agree it might be seen as breaking but looking at the problem description I think the previous change was accidentally breaking (i.e. changing from effectively infinite to finite) and this one is simply a bugfix restoring it.

Then I can accept this as non breaking change but a bug fix. Yet requesting to add a note explaining it in the top of the CHANGELOG. I will bake it to the proper release version during release process.

[mjsqu]

Changelog note added 456ee50

Unsure of the best way to refer back to the original PR that I am partially reverting, I trust you'll properly format the Changelog message when it comes to release time. TIA.

[eladkal]

Lets modify this to a more explanatory text.
Dont mention PR numbers. The commit title that we add during release cycle will do that and refer to this PR for more details.

What we are seeking in the note is just a simple explnatipn of the situation "A bug intoduced in provider version... Caused.... In this version we are fixing it by returning value to ..."
Something like this.

[eladkal]

LGTM

needs rebase and resolve conflicts

[mjsqu]

LGTM

needs rebase and resolve conflicts

Resolved conflict in CHANGELOG.rst - 56782d5

[eladkal]

Tests fail :(

[mjsqu]

Tests fail :(

Tests were failing due to the size of sys.maxsize. The waiter_max_attempts value is used in a timedelta calculation here (https://github.com/mjsqu/airflow/blob/b9453e9674e69203ca8658aa15f7e5c24e88f4db/airflow/providers/amazon/aws/operators/ecs.py#L564)

>>>timedelta(seconds=sys.maxsize*delay+60)
Traceback (mostrecentcalllast):
File"<stdin>", line1, in<module>OverflowError: PythoninttoolargetoconverttoCint

To resolve this, I have set the default waiter_max_attempts value high enough such that the waiter would run for 1M years. This allows the timedelta calculation to succeed, while providing a value that is likely longer than any ECS Tasks will ever run for.

max_attempts=1000000*365*24*60*10>>>timedelta(seconds=max_attempts*delay+60)
datetime.timedelta(days=365000000, seconds=60)

Alternatively, if possible, it would be good to calculate a default_max_attempts at runtime using execution_timeout, I'm just unsure how.

I'm open to suggestions about other values to use instead of 1 million years, there are lower values that would be equally sensible.

[mjsqu]

Latest test failures:

cluster="c", tasks=["arn"], WaiterConfig={"Delay": 6, "MaxAttempts": 1000000*365*24*60*10}
)
>assert1000000*365*24*60*10==client_mock.get_waiter.return_value.config.max_attemptsEAssertionError: assert ((((1000000*365) *24) *60) *10) ==<MagicMockname='client.get_waiter().config.max_attempts'id='139852189801392'>E+where<MagicMockname='client.get_waiter().config.max_attempts'id='139852189801392'>=<MagicMockname='client.get_waiter().config'id='139852190309584'>.max_attemptsE+where<MagicMockname='client.get_waiter().config'id='139852190309584'>=<MagicMockname='client.get_waiter()'id='139852190269200'>.configE+where<MagicMockname='client.get_waiter()'id='139852190269200'>=<MagicMockname='client.get_waiter'id='139852190252720'>.return_valueE+where<MagicMockname='client.get_waiter'id='139852190252720'>=<MagicMockname='client'id='139852190228688'>.get_waiter

Struggling to understand how the assertion fails, perhaps there's a precision issue such that client_mock.get_waiter.return_value.config.max_attempts does not return the exact value it was provided with?

[mjsqu]

Anyone following this PR/Issue, we have set the waiter_max_attempts value to sys.maxsize in the interim for affected EcsRunTaskOperator tasks, but you could choose a sufficiently large value to force the waiter to run for longer than execution_timeout:

waiter_max_attempts * waiter_delay > execution_timeout

e.g. for an execution_timeout of 1 hour (3600 seconds), set:

waiter_max_attempts=650, # Greater than 600waiter_delay=6,
# Total waiter time = 3,900 seconds == 65 minutes

[hussein-awala]

The failed check was not valid after removing this line.

[vincbeck]

Thanks for fixing that bug!

[boring-cyborg]

Awesome work, congrats on your first merged pull request! You are invited to check our Issue Tracker for additional contributions.

[mjsqu]

---

closes: #33711
related: #33016, #30586

Prior to v8.0.0 the default waiter_max_attempts parameter in EcsRunTaskOperator was set to sys.maxsize. The value is a large integer, which should always exceed the execution_timeout value for a given task. I believe this is why the line setting the value to sys.maxsize is accompanied by a comment:

waiter.config.max_attempts=sys.maxsize# timeout is managed by airflow

Changes in v8.0.0 introduced a configurable value defaulting to 100 for this parameter, with a delay defaulting to 6 seconds (the changes were introduced to resolve#30586). The line above remains, but I think it is being overridden by the call to waiter.wait:

waiter.wait(
cluster=self.cluster,
tasks=[self.arn],
WaiterConfig={
"Delay": self.waiter_delay,
"MaxAttempts": self.waiter_max_attempts, # !!! Overriding previously set max_attempts?
},
)

The defaults for these previously unavailable parameters have been set to:

waiter_delay: int=6,
waiter_max_attempts: int=100,

I believe the introduction of a default value of 100 has changed the behaviour of the operator - now when not specified, all tasks detach from the running ECS Task at 10 minutes, reporting failed status back to Airflow.

The fix replaces the default value with sys.maxsize - which should revert the waiter functionality back to prior working versions.

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

[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 Contribution Guide (https://github.com/apache/airflow/blob/main/CONTRIBUTING.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: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[mjsqu]

While I wait for my sleep 800 command to run having manually set waiter_max_attempts to sys.maxsize...

Here's the documentation for the ECS TasksStopped waiter: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ecs/waiter/TasksStopped.html

If not set, it has defaults of

WaiterConfig={
'Delay': 6,
'MaxAttempts': 100
}

too. I think this was the motivation for setting these as the defaults, however the prior functionality would override the MaxAttempts to a really high number and let Airflow's execution_timeout parameter handle long-running tasks. The execution_timeout method works well, when the Airflow timeout is hit, the ECS task is killed.

[hussein-awala]

There are two things here:

1. Changing the default value to infinite could be a breaking change (cc: @eladkal )
2. If it's okay to change it, the best way is to make it None and replace the loop in the Trigger with while True when it's None.

[potiuk]

Agree it might be seen as breaking but looking at the problem description I think the previous change was accidentally breaking (i.e. changing from effectively infinite to finite) and this one is simply a bugfix restoring it.

[mjsqu]

Thanks for the review. I agree that adding sys.maxsize must have been a workaround and is a bit non-standard so perhaps this PR might be able to resolve that by using some other method.

I had considered if the max_attempts and delay parameters could be calculated at runtime by taking into consideration the execution_timeout value, however this would require a bit more of a rewrite of the Operator.

[eladkal]

Agree it might be seen as breaking but looking at the problem description I think the previous change was accidentally breaking (i.e. changing from effectively infinite to finite) and this one is simply a bugfix restoring it.

Then I can accept this as non breaking change but a bug fix. Yet requesting to add a note explaining it in the top of the CHANGELOG. I will bake it to the proper release version during release process.

[mjsqu]

Changelog note added 456ee50

Unsure of the best way to refer back to the original PR that I am partially reverting, I trust you'll properly format the Changelog message when it comes to release time. TIA.

[eladkal]

Lets modify this to a more explanatory text.
Dont mention PR numbers. The commit title that we add during release cycle will do that and refer to this PR for more details.

What we are seeking in the note is just a simple explnatipn of the situation "A bug intoduced in provider version... Caused.... In this version we are fixing it by returning value to ..."
Something like this.

[eladkal]

LGTM

needs rebase and resolve conflicts

[mjsqu]

LGTM

needs rebase and resolve conflicts

Resolved conflict in CHANGELOG.rst - 56782d5

[eladkal]

Tests fail :(

[mjsqu]

Tests fail :(

Tests were failing due to the size of sys.maxsize. The waiter_max_attempts value is used in a timedelta calculation here (https://github.com/mjsqu/airflow/blob/b9453e9674e69203ca8658aa15f7e5c24e88f4db/airflow/providers/amazon/aws/operators/ecs.py#L564)

>>>timedelta(seconds=sys.maxsize*delay+60)
Traceback (mostrecentcalllast):
File"<stdin>", line1, in<module>OverflowError: PythoninttoolargetoconverttoCint

To resolve this, I have set the default waiter_max_attempts value high enough such that the waiter would run for 1M years. This allows the timedelta calculation to succeed, while providing a value that is likely longer than any ECS Tasks will ever run for.

max_attempts=1000000*365*24*60*10>>>timedelta(seconds=max_attempts*delay+60)
datetime.timedelta(days=365000000, seconds=60)

Alternatively, if possible, it would be good to calculate a default_max_attempts at runtime using execution_timeout, I'm just unsure how.

I'm open to suggestions about other values to use instead of 1 million years, there are lower values that would be equally sensible.

[mjsqu]

Latest test failures:

cluster="c", tasks=["arn"], WaiterConfig={"Delay": 6, "MaxAttempts": 1000000*365*24*60*10}
)
>assert1000000*365*24*60*10==client_mock.get_waiter.return_value.config.max_attemptsEAssertionError: assert ((((1000000*365) *24) *60) *10) ==<MagicMockname='client.get_waiter().config.max_attempts'id='139852189801392'>E+where<MagicMockname='client.get_waiter().config.max_attempts'id='139852189801392'>=<MagicMockname='client.get_waiter().config'id='139852190309584'>.max_attemptsE+where<MagicMockname='client.get_waiter().config'id='139852190309584'>=<MagicMockname='client.get_waiter()'id='139852190269200'>.configE+where<MagicMockname='client.get_waiter()'id='139852190269200'>=<MagicMockname='client.get_waiter'id='139852190252720'>.return_valueE+where<MagicMockname='client.get_waiter'id='139852190252720'>=<MagicMockname='client'id='139852190228688'>.get_waiter

Struggling to understand how the assertion fails, perhaps there's a precision issue such that client_mock.get_waiter.return_value.config.max_attempts does not return the exact value it was provided with?

[mjsqu]

Anyone following this PR/Issue, we have set the waiter_max_attempts value to sys.maxsize in the interim for affected EcsRunTaskOperator tasks, but you could choose a sufficiently large value to force the waiter to run for longer than execution_timeout:

waiter_max_attempts * waiter_delay > execution_timeout

e.g. for an execution_timeout of 1 hour (3600 seconds), set:

waiter_max_attempts=650, # Greater than 600waiter_delay=6,
# Total waiter time = 3,900 seconds == 65 minutes

[hussein-awala]

The failed check was not valid after removing this line.

[vincbeck]

Thanks for fixing that bug!

[boring-cyborg]

Awesome work, congrats on your first merged pull request! You are invited to check our Issue Tracker for additional contributions.

[mjsqu]

---

closes: #33711
related: #33016, #30586

Prior to v8.0.0 the default waiter_max_attempts parameter in EcsRunTaskOperator was set to sys.maxsize. The value is a large integer, which should always exceed the execution_timeout value for a given task. I believe this is why the line setting the value to sys.maxsize is accompanied by a comment:

waiter.config.max_attempts=sys.maxsize# timeout is managed by airflow

Changes in v8.0.0 introduced a configurable value defaulting to 100 for this parameter, with a delay defaulting to 6 seconds (the changes were introduced to resolve#30586). The line above remains, but I think it is being overridden by the call to waiter.wait:

waiter.wait(
cluster=self.cluster,
tasks=[self.arn],
WaiterConfig={
"Delay": self.waiter_delay,
"MaxAttempts": self.waiter_max_attempts, # !!! Overriding previously set max_attempts?
},
)

The defaults for these previously unavailable parameters have been set to:

waiter_delay: int=6,
waiter_max_attempts: int=100,

I believe the introduction of a default value of 100 has changed the behaviour of the operator - now when not specified, all tasks detach from the running ECS Task at 10 minutes, reporting failed status back to Airflow.

The fix replaces the default value with sys.maxsize - which should revert the waiter functionality back to prior working versions.

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

[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 Contribution Guide (https://github.com/apache/airflow/blob/main/CONTRIBUTING.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: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[mjsqu]

While I wait for my sleep 800 command to run having manually set waiter_max_attempts to sys.maxsize...

Here's the documentation for the ECS TasksStopped waiter: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ecs/waiter/TasksStopped.html

If not set, it has defaults of

WaiterConfig={
'Delay': 6,
'MaxAttempts': 100
}

too. I think this was the motivation for setting these as the defaults, however the prior functionality would override the MaxAttempts to a really high number and let Airflow's execution_timeout parameter handle long-running tasks. The execution_timeout method works well, when the Airflow timeout is hit, the ECS task is killed.

[hussein-awala]

There are two things here:

1. Changing the default value to infinite could be a breaking change (cc: @eladkal )
2. If it's okay to change it, the best way is to make it None and replace the loop in the Trigger with while True when it's None.

[potiuk]

Agree it might be seen as breaking but looking at the problem description I think the previous change was accidentally breaking (i.e. changing from effectively infinite to finite) and this one is simply a bugfix restoring it.

[mjsqu]

Thanks for the review. I agree that adding sys.maxsize must have been a workaround and is a bit non-standard so perhaps this PR might be able to resolve that by using some other method.

I had considered if the max_attempts and delay parameters could be calculated at runtime by taking into consideration the execution_timeout value, however this would require a bit more of a rewrite of the Operator.

[eladkal]

Agree it might be seen as breaking but looking at the problem description I think the previous change was accidentally breaking (i.e. changing from effectively infinite to finite) and this one is simply a bugfix restoring it.

Then I can accept this as non breaking change but a bug fix. Yet requesting to add a note explaining it in the top of the CHANGELOG. I will bake it to the proper release version during release process.

[mjsqu]

Changelog note added 456ee50

Unsure of the best way to refer back to the original PR that I am partially reverting, I trust you'll properly format the Changelog message when it comes to release time. TIA.

[eladkal]

Lets modify this to a more explanatory text.
Dont mention PR numbers. The commit title that we add during release cycle will do that and refer to this PR for more details.

What we are seeking in the note is just a simple explnatipn of the situation "A bug intoduced in provider version... Caused.... In this version we are fixing it by returning value to ..."
Something like this.

[eladkal]

LGTM

needs rebase and resolve conflicts

[mjsqu]

LGTM

needs rebase and resolve conflicts

Resolved conflict in CHANGELOG.rst - 56782d5

[eladkal]

Tests fail :(

[mjsqu]

Tests fail :(

Tests were failing due to the size of sys.maxsize. The waiter_max_attempts value is used in a timedelta calculation here (https://github.com/mjsqu/airflow/blob/b9453e9674e69203ca8658aa15f7e5c24e88f4db/airflow/providers/amazon/aws/operators/ecs.py#L564)

>>>timedelta(seconds=sys.maxsize*delay+60)
Traceback (mostrecentcalllast):
File"<stdin>", line1, in<module>OverflowError: PythoninttoolargetoconverttoCint

To resolve this, I have set the default waiter_max_attempts value high enough such that the waiter would run for 1M years. This allows the timedelta calculation to succeed, while providing a value that is likely longer than any ECS Tasks will ever run for.

max_attempts=1000000*365*24*60*10>>>timedelta(seconds=max_attempts*delay+60)
datetime.timedelta(days=365000000, seconds=60)

Alternatively, if possible, it would be good to calculate a default_max_attempts at runtime using execution_timeout, I'm just unsure how.

I'm open to suggestions about other values to use instead of 1 million years, there are lower values that would be equally sensible.

[mjsqu]

Latest test failures:

cluster="c", tasks=["arn"], WaiterConfig={"Delay": 6, "MaxAttempts": 1000000*365*24*60*10}
)
>assert1000000*365*24*60*10==client_mock.get_waiter.return_value.config.max_attemptsEAssertionError: assert ((((1000000*365) *24) *60) *10) ==<MagicMockname='client.get_waiter().config.max_attempts'id='139852189801392'>E+where<MagicMockname='client.get_waiter().config.max_attempts'id='139852189801392'>=<MagicMockname='client.get_waiter().config'id='139852190309584'>.max_attemptsE+where<MagicMockname='client.get_waiter().config'id='139852190309584'>=<MagicMockname='client.get_waiter()'id='139852190269200'>.configE+where<MagicMockname='client.get_waiter()'id='139852190269200'>=<MagicMockname='client.get_waiter'id='139852190252720'>.return_valueE+where<MagicMockname='client.get_waiter'id='139852190252720'>=<MagicMockname='client'id='139852190228688'>.get_waiter

Struggling to understand how the assertion fails, perhaps there's a precision issue such that client_mock.get_waiter.return_value.config.max_attempts does not return the exact value it was provided with?

[mjsqu]

Anyone following this PR/Issue, we have set the waiter_max_attempts value to sys.maxsize in the interim for affected EcsRunTaskOperator tasks, but you could choose a sufficiently large value to force the waiter to run for longer than execution_timeout:

waiter_max_attempts * waiter_delay > execution_timeout

e.g. for an execution_timeout of 1 hour (3600 seconds), set:

waiter_max_attempts=650, # Greater than 600waiter_delay=6,
# Total waiter time = 3,900 seconds == 65 minutes

[hussein-awala]

The failed check was not valid after removing this line.

[vincbeck]

Thanks for fixing that bug!

[boring-cyborg]

Awesome work, congrats on your first merged pull request! You are invited to check our Issue Tracker for additional contributions.

[mjsqu]

---

closes: #33711
related: #33016, #30586

Prior to v8.0.0 the default waiter_max_attempts parameter in EcsRunTaskOperator was set to sys.maxsize. The value is a large integer, which should always exceed the execution_timeout value for a given task. I believe this is why the line setting the value to sys.maxsize is accompanied by a comment:

waiter.config.max_attempts=sys.maxsize# timeout is managed by airflow

Changes in v8.0.0 introduced a configurable value defaulting to 100 for this parameter, with a delay defaulting to 6 seconds (the changes were introduced to resolve#30586). The line above remains, but I think it is being overridden by the call to waiter.wait:

waiter.wait(
cluster=self.cluster,
tasks=[self.arn],
WaiterConfig={
"Delay": self.waiter_delay,
"MaxAttempts": self.waiter_max_attempts, # !!! Overriding previously set max_attempts?
},
)

The defaults for these previously unavailable parameters have been set to:

waiter_delay: int=6,
waiter_max_attempts: int=100,

I believe the introduction of a default value of 100 has changed the behaviour of the operator - now when not specified, all tasks detach from the running ECS Task at 10 minutes, reporting failed status back to Airflow.

The fix replaces the default value with sys.maxsize - which should revert the waiter functionality back to prior working versions.

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

[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 Contribution Guide (https://github.com/apache/airflow/blob/main/CONTRIBUTING.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: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[mjsqu]

While I wait for my sleep 800 command to run having manually set waiter_max_attempts to sys.maxsize...

Here's the documentation for the ECS TasksStopped waiter: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ecs/waiter/TasksStopped.html

If not set, it has defaults of

WaiterConfig={
'Delay': 6,
'MaxAttempts': 100
}

too. I think this was the motivation for setting these as the defaults, however the prior functionality would override the MaxAttempts to a really high number and let Airflow's execution_timeout parameter handle long-running tasks. The execution_timeout method works well, when the Airflow timeout is hit, the ECS task is killed.

[hussein-awala]

There are two things here:

1. Changing the default value to infinite could be a breaking change (cc: @eladkal )
2. If it's okay to change it, the best way is to make it None and replace the loop in the Trigger with while True when it's None.

[potiuk]

Agree it might be seen as breaking but looking at the problem description I think the previous change was accidentally breaking (i.e. changing from effectively infinite to finite) and this one is simply a bugfix restoring it.

[mjsqu]

Thanks for the review. I agree that adding sys.maxsize must have been a workaround and is a bit non-standard so perhaps this PR might be able to resolve that by using some other method.

I had considered if the max_attempts and delay parameters could be calculated at runtime by taking into consideration the execution_timeout value, however this would require a bit more of a rewrite of the Operator.

[eladkal]

Agree it might be seen as breaking but looking at the problem description I think the previous change was accidentally breaking (i.e. changing from effectively infinite to finite) and this one is simply a bugfix restoring it.

Then I can accept this as non breaking change but a bug fix. Yet requesting to add a note explaining it in the top of the CHANGELOG. I will bake it to the proper release version during release process.

[mjsqu]

Changelog note added 456ee50

Unsure of the best way to refer back to the original PR that I am partially reverting, I trust you'll properly format the Changelog message when it comes to release time. TIA.

[eladkal]

Lets modify this to a more explanatory text.
Dont mention PR numbers. The commit title that we add during release cycle will do that and refer to this PR for more details.

What we are seeking in the note is just a simple explnatipn of the situation "A bug intoduced in provider version... Caused.... In this version we are fixing it by returning value to ..."
Something like this.

[eladkal]

LGTM

needs rebase and resolve conflicts

[mjsqu]

LGTM

needs rebase and resolve conflicts

Resolved conflict in CHANGELOG.rst - 56782d5

[eladkal]

Tests fail :(

[mjsqu]

Tests fail :(

Tests were failing due to the size of sys.maxsize. The waiter_max_attempts value is used in a timedelta calculation here (https://github.com/mjsqu/airflow/blob/b9453e9674e69203ca8658aa15f7e5c24e88f4db/airflow/providers/amazon/aws/operators/ecs.py#L564)

>>>timedelta(seconds=sys.maxsize*delay+60)
Traceback (mostrecentcalllast):
File"<stdin>", line1, in<module>OverflowError: PythoninttoolargetoconverttoCint

To resolve this, I have set the default waiter_max_attempts value high enough such that the waiter would run for 1M years. This allows the timedelta calculation to succeed, while providing a value that is likely longer than any ECS Tasks will ever run for.

max_attempts=1000000*365*24*60*10>>>timedelta(seconds=max_attempts*delay+60)
datetime.timedelta(days=365000000, seconds=60)

Alternatively, if possible, it would be good to calculate a default_max_attempts at runtime using execution_timeout, I'm just unsure how.

I'm open to suggestions about other values to use instead of 1 million years, there are lower values that would be equally sensible.

[mjsqu]

Latest test failures:

cluster="c", tasks=["arn"], WaiterConfig={"Delay": 6, "MaxAttempts": 1000000*365*24*60*10}
)
>assert1000000*365*24*60*10==client_mock.get_waiter.return_value.config.max_attemptsEAssertionError: assert ((((1000000*365) *24) *60) *10) ==<MagicMockname='client.get_waiter().config.max_attempts'id='139852189801392'>E+where<MagicMockname='client.get_waiter().config.max_attempts'id='139852189801392'>=<MagicMockname='client.get_waiter().config'id='139852190309584'>.max_attemptsE+where<MagicMockname='client.get_waiter().config'id='139852190309584'>=<MagicMockname='client.get_waiter()'id='139852190269200'>.configE+where<MagicMockname='client.get_waiter()'id='139852190269200'>=<MagicMockname='client.get_waiter'id='139852190252720'>.return_valueE+where<MagicMockname='client.get_waiter'id='139852190252720'>=<MagicMockname='client'id='139852190228688'>.get_waiter

Struggling to understand how the assertion fails, perhaps there's a precision issue such that client_mock.get_waiter.return_value.config.max_attempts does not return the exact value it was provided with?

[mjsqu]

Anyone following this PR/Issue, we have set the waiter_max_attempts value to sys.maxsize in the interim for affected EcsRunTaskOperator tasks, but you could choose a sufficiently large value to force the waiter to run for longer than execution_timeout:

waiter_max_attempts * waiter_delay > execution_timeout

e.g. for an execution_timeout of 1 hour (3600 seconds), set:

waiter_max_attempts=650, # Greater than 600waiter_delay=6,
# Total waiter time = 3,900 seconds == 65 minutes

[hussein-awala]

The failed check was not valid after removing this line.

[vincbeck]

Thanks for fixing that bug!

[boring-cyborg]

Awesome work, congrats on your first merged pull request! You are invited to check our Issue Tracker for additional contributions.

[mjsqu]

---

closes: #33711
related: #33016, #30586

Prior to v8.0.0 the default waiter_max_attempts parameter in EcsRunTaskOperator was set to sys.maxsize. The value is a large integer, which should always exceed the execution_timeout value for a given task. I believe this is why the line setting the value to sys.maxsize is accompanied by a comment:

waiter.config.max_attempts=sys.maxsize# timeout is managed by airflow

Changes in v8.0.0 introduced a configurable value defaulting to 100 for this parameter, with a delay defaulting to 6 seconds (the changes were introduced to resolve#30586). The line above remains, but I think it is being overridden by the call to waiter.wait:

waiter.wait(
cluster=self.cluster,
tasks=[self.arn],
WaiterConfig={
"Delay": self.waiter_delay,
"MaxAttempts": self.waiter_max_attempts, # !!! Overriding previously set max_attempts?
},
)

The defaults for these previously unavailable parameters have been set to:

waiter_delay: int=6,
waiter_max_attempts: int=100,

I believe the introduction of a default value of 100 has changed the behaviour of the operator - now when not specified, all tasks detach from the running ECS Task at 10 minutes, reporting failed status back to Airflow.

The fix replaces the default value with sys.maxsize - which should revert the waiter functionality back to prior working versions.

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

[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 Contribution Guide (https://github.com/apache/airflow/blob/main/CONTRIBUTING.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: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[mjsqu]

While I wait for my sleep 800 command to run having manually set waiter_max_attempts to sys.maxsize...

Here's the documentation for the ECS TasksStopped waiter: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ecs/waiter/TasksStopped.html

If not set, it has defaults of

WaiterConfig={
'Delay': 6,
'MaxAttempts': 100
}

too. I think this was the motivation for setting these as the defaults, however the prior functionality would override the MaxAttempts to a really high number and let Airflow's execution_timeout parameter handle long-running tasks. The execution_timeout method works well, when the Airflow timeout is hit, the ECS task is killed.

[hussein-awala]

There are two things here:

1. Changing the default value to infinite could be a breaking change (cc: @eladkal )
2. If it's okay to change it, the best way is to make it None and replace the loop in the Trigger with while True when it's None.

[potiuk]

Agree it might be seen as breaking but looking at the problem description I think the previous change was accidentally breaking (i.e. changing from effectively infinite to finite) and this one is simply a bugfix restoring it.

[mjsqu]

Thanks for the review. I agree that adding sys.maxsize must have been a workaround and is a bit non-standard so perhaps this PR might be able to resolve that by using some other method.

I had considered if the max_attempts and delay parameters could be calculated at runtime by taking into consideration the execution_timeout value, however this would require a bit more of a rewrite of the Operator.

[eladkal]

Agree it might be seen as breaking but looking at the problem description I think the previous change was accidentally breaking (i.e. changing from effectively infinite to finite) and this one is simply a bugfix restoring it.

Then I can accept this as non breaking change but a bug fix. Yet requesting to add a note explaining it in the top of the CHANGELOG. I will bake it to the proper release version during release process.

[mjsqu]

Changelog note added 456ee50

Unsure of the best way to refer back to the original PR that I am partially reverting, I trust you'll properly format the Changelog message when it comes to release time. TIA.

[eladkal]

Lets modify this to a more explanatory text.
Dont mention PR numbers. The commit title that we add during release cycle will do that and refer to this PR for more details.

What we are seeking in the note is just a simple explnatipn of the situation "A bug intoduced in provider version... Caused.... In this version we are fixing it by returning value to ..."
Something like this.

[eladkal]

LGTM

needs rebase and resolve conflicts

[mjsqu]

LGTM

needs rebase and resolve conflicts

Resolved conflict in CHANGELOG.rst - 56782d5

[eladkal]

Tests fail :(

[mjsqu]

Tests fail :(

Tests were failing due to the size of sys.maxsize. The waiter_max_attempts value is used in a timedelta calculation here (https://github.com/mjsqu/airflow/blob/b9453e9674e69203ca8658aa15f7e5c24e88f4db/airflow/providers/amazon/aws/operators/ecs.py#L564)

>>>timedelta(seconds=sys.maxsize*delay+60)
Traceback (mostrecentcalllast):
File"<stdin>", line1, in<module>OverflowError: PythoninttoolargetoconverttoCint

To resolve this, I have set the default waiter_max_attempts value high enough such that the waiter would run for 1M years. This allows the timedelta calculation to succeed, while providing a value that is likely longer than any ECS Tasks will ever run for.

max_attempts=1000000*365*24*60*10>>>timedelta(seconds=max_attempts*delay+60)
datetime.timedelta(days=365000000, seconds=60)

Alternatively, if possible, it would be good to calculate a default_max_attempts at runtime using execution_timeout, I'm just unsure how.

I'm open to suggestions about other values to use instead of 1 million years, there are lower values that would be equally sensible.

[mjsqu]

Latest test failures:

cluster="c", tasks=["arn"], WaiterConfig={"Delay": 6, "MaxAttempts": 1000000*365*24*60*10}
)
>assert1000000*365*24*60*10==client_mock.get_waiter.return_value.config.max_attemptsEAssertionError: assert ((((1000000*365) *24) *60) *10) ==<MagicMockname='client.get_waiter().config.max_attempts'id='139852189801392'>E+where<MagicMockname='client.get_waiter().config.max_attempts'id='139852189801392'>=<MagicMockname='client.get_waiter().config'id='139852190309584'>.max_attemptsE+where<MagicMockname='client.get_waiter().config'id='139852190309584'>=<MagicMockname='client.get_waiter()'id='139852190269200'>.configE+where<MagicMockname='client.get_waiter()'id='139852190269200'>=<MagicMockname='client.get_waiter'id='139852190252720'>.return_valueE+where<MagicMockname='client.get_waiter'id='139852190252720'>=<MagicMockname='client'id='139852190228688'>.get_waiter

Struggling to understand how the assertion fails, perhaps there's a precision issue such that client_mock.get_waiter.return_value.config.max_attempts does not return the exact value it was provided with?

[mjsqu]

Anyone following this PR/Issue, we have set the waiter_max_attempts value to sys.maxsize in the interim for affected EcsRunTaskOperator tasks, but you could choose a sufficiently large value to force the waiter to run for longer than execution_timeout:

waiter_max_attempts * waiter_delay > execution_timeout

e.g. for an execution_timeout of 1 hour (3600 seconds), set:

waiter_max_attempts=650, # Greater than 600waiter_delay=6,
# Total waiter time = 3,900 seconds == 65 minutes

[hussein-awala]

The failed check was not valid after removing this line.

[vincbeck]

Thanks for fixing that bug!

[boring-cyborg]

Awesome work, congrats on your first merged pull request! You are invited to check our Issue Tracker for additional contributions.