diff --git a/airflow/providers/amazon/CHANGELOG.rst b/airflow/providers/amazon/CHANGELOG.rst index 3a61e959d88a9..b937a158ec822 100644 --- a/airflow/providers/amazon/CHANGELOG.rst +++ b/airflow/providers/amazon/CHANGELOG.rst @@ -25,6 +25,7 @@ Changelog --------- +* ``A bug intoduced in provider-amazon version 8.0.0 caused all 'EcsRunTaskOperator' tasks to detach from the ECS task after 10 minutes and fail - even if the ECS task was still running. In this version we are fixing it by returning the default 'waiter_max_attempts' value to 'sys.maxsize'`` 8.6.0 ..... diff --git a/airflow/providers/amazon/aws/operators/ecs.py b/airflow/providers/amazon/aws/operators/ecs.py index ed8e5ccec35f8..ad500db40a30b 100644 --- a/airflow/providers/amazon/aws/operators/ecs.py +++ b/airflow/providers/amazon/aws/operators/ecs.py @@ -18,7 +18,6 @@ from __future__ import annotations import re -import sys import warnings from datetime import timedelta from functools import cached_property @@ -476,7 +475,9 @@ def __init__( number_logs_exception: int = 10, wait_for_completion: bool = True, waiter_delay: int = 6, - waiter_max_attempts: int = 100, + waiter_max_attempts: int = 1000000 * 365 * 24 * 60 * 10, + # Set the default waiter duration to 1M years (attempts*delay) + # Airflow execution_timeout handles task timeout deferrable: bool = conf.getboolean("operators", "default_deferrable", fallback=False), **kwargs, ): @@ -665,7 +666,6 @@ def _wait_for_task_ended(self) -> None: return waiter = self.client.get_waiter("tasks_stopped") - waiter.config.max_attempts = sys.maxsize # timeout is managed by airflow waiter.wait( cluster=self.cluster, tasks=[self.arn], diff --git a/tests/providers/amazon/aws/operators/test_ecs.py b/tests/providers/amazon/aws/operators/test_ecs.py index 9fad66ec2ad31..c3a33ab47fe90 100644 --- a/tests/providers/amazon/aws/operators/test_ecs.py +++ b/tests/providers/amazon/aws/operators/test_ecs.py @@ -17,7 +17,6 @@ # under the License. from __future__ import annotations -import sys from copy import deepcopy from unittest import mock from unittest.mock import MagicMock, PropertyMock @@ -348,9 +347,8 @@ def test_wait_end_tasks(self, client_mock): self.ecs._wait_for_task_ended() client_mock.get_waiter.assert_called_once_with("tasks_stopped") client_mock.get_waiter.return_value.wait.assert_called_once_with( - cluster="c", tasks=["arn"], WaiterConfig={"Delay": 6, "MaxAttempts": 100} + cluster="c", tasks=["arn"], WaiterConfig={"Delay": 6, "MaxAttempts": 1000000 * 365 * 24 * 60 * 10} ) - assert sys.maxsize == client_mock.get_waiter.return_value.config.max_attempts @mock.patch.object(EcsBaseOperator, "client") def test_check_success_tasks_raises_failed_to_start(self, client_mock):