Fix ecs/EcsRunTaskOperator retry condition

providers/amazon/src/airflow/providers/amazon/aws/exceptions.py

@@ -48,5 +48,16 @@ def __reduce__(self):
         return EcsOperatorError, (self.failures, self.message)
 
 
+class EcsCannotPullContainerError(Exception):
+    """Raise when ECS cannot retrieve the specified container image."""
+
+    def __init__(self, message: str):
+        self.message = message
+        super().__init__(message)
+
+    def __reduce__(self):
+        return self.__class__, (self.message,)
+
+
 class S3HookUriParseFailure(AirflowException):
     """When parse_s3_url fails to parse URL, this error is thrown."""

providers/amazon/src/airflow/providers/amazon/aws/hooks/ecs.py

@@ -19,7 +19,11 @@
 
 from typing import TYPE_CHECKING, Protocol, runtime_checkable
 
-from airflow.providers.amazon.aws.exceptions import EcsOperatorError, EcsTaskFailToStart
+from airflow.providers.amazon.aws.exceptions import (
+    EcsCannotPullContainerError,
+    EcsOperatorError,
+    EcsTaskFailToStart,
+)
 from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
 from airflow.providers.amazon.aws.utils import _StringCompareEnum
 
@@ -29,6 +33,9 @@
 
 def should_retry(exception: Exception):
     """Check if exception is related to ECS resource quota (CPU, MEM)."""
+    if isinstance(exception, EcsCannotPullContainerError):
+        return False
+
     if isinstance(exception, EcsOperatorError):
         return any(
             quota_reason in failure["reason"]
@@ -40,6 +47,9 @@ def should_retry(exception: Exception):
 
 def should_retry_eni(exception: Exception):
     """Check if exception is related to ENI (Elastic Network Interfaces)."""
+    if isinstance(exception, EcsCannotPullContainerError):
+        return False
+
     if isinstance(exception, EcsTaskFailToStart):
         return any(
             eni_reason in exception.message

providers/amazon/src/airflow/providers/amazon/aws/operators/ecs.py

@@ -25,7 +25,11 @@
 
 from airflow.configuration import conf
 from airflow.exceptions import AirflowException
-from airflow.providers.amazon.aws.exceptions import EcsOperatorError, EcsTaskFailToStart
+from airflow.providers.amazon.aws.exceptions import (
+    EcsCannotPullContainerError,
+    EcsOperatorError,
+    EcsTaskFailToStart,
+)
 from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
 from airflow.providers.amazon.aws.hooks.ecs import EcsClusterStates, EcsHook, should_retry_eni
 from airflow.providers.amazon.aws.hooks.logs import AwsLogsHook
@@ -701,6 +705,11 @@ def _check_success_task(self) -> None:
 
         for task in response["tasks"]:
             if task.get("stopCode", "") == "TaskFailedToStart":
+                if "CannotPullContainerError" in task.get("stoppedReason", ""):
+                    raise EcsCannotPullContainerError(
+                        f"The task failed to start due to: {task.get('stoppedReason', '')}"
+                    )
+
                 # Reset task arn here otherwise the retry run will not start
                 # a new task but keep polling the old dead one
                 # I'm not resetting it for other exceptions here because

providers/amazon/tests/unit/amazon/aws/hooks/test_ecs.py

@@ -20,7 +20,11 @@
 
 import pytest
 
-from airflow.providers.amazon.aws.exceptions import EcsOperatorError, EcsTaskFailToStart
+from airflow.providers.amazon.aws.exceptions import (
+    EcsCannotPullContainerError,
+    EcsOperatorError,
+    EcsTaskFailToStart,
+)
 from airflow.providers.amazon.aws.hooks.ecs import EcsHook, should_retry, should_retry_eni
 
 DEFAULT_CONN_ID: str = "aws_default"
@@ -69,6 +73,13 @@ def test_return_true_on_valid_reason(self):
                 "Timeout waiting for network interface provisioning to complete."
             )
         )
+        assert should_retry_eni(
+            EcsCannotPullContainerError(
+                "The task failed to start due to: "
+                "CannotStartContainerError: "
+                "ResourceInitializationError: failed to create new container runtime task"
+            )
+        )
 
     def test_return_false_on_invalid_reason(self):
         assert not should_retry_eni(

providers/amazon/tests/unit/amazon/aws/operators/test_ecs.py

@@ -25,7 +25,11 @@
 import pytest
 
 from airflow.exceptions import AirflowException, TaskDeferred
-from airflow.providers.amazon.aws.exceptions import EcsOperatorError, EcsTaskFailToStart
+from airflow.providers.amazon.aws.exceptions import (
+    EcsCannotPullContainerError,
+    EcsOperatorError,
+    EcsTaskFailToStart,
+)
 from airflow.providers.amazon.aws.hooks.ecs import EcsClusterStates, EcsHook
 from airflow.providers.amazon.aws.operators.ecs import (
     EcsBaseOperator,
@@ -444,6 +448,28 @@ def test_check_success_tasks_raises_failed_to_start(self, client_mock):
         assert str(ctx.value) == "The task failed to start due to: Task failed to start"
         client_mock.describe_tasks.assert_called_once_with(cluster="c", tasks=["arn"])
 
+    @mock.patch.object(EcsBaseOperator, "client")
+    def test_check_success_tasks_raises_cannot_pull_container_error(self, client_mock):
+        self.ecs.arn = "arn"
+        client_mock.describe_tasks.return_value = {
+            "tasks": [
+                {
+                    "stopCode": "TaskFailedToStart",
+                    "stoppedReason": "CannotPullContainerError: ResourceInitializationError: failed to create new container runtime task",
+                    "containers": [{"name": "foo", "lastStatus": "STOPPED"}],
+                }
+            ]
+        }
+
+        with pytest.raises(EcsCannotPullContainerError) as ctx:
+            self.ecs._check_success_task()
+
+        assert (
+            str(ctx.value)
+            == "The task failed to start due to: CannotPullContainerError: ResourceInitializationError: failed to create new container runtime task"
+        )
+        client_mock.describe_tasks.assert_called_once_with(cluster="c", tasks=["arn"])
+
     @mock.patch.object(EcsBaseOperator, "client")
     @mock.patch("airflow.providers.amazon.aws.utils.task_log_fetcher.AwsTaskLogFetcher")
     def test_check_success_tasks_raises_cloudwatch_logs(self, log_fetcher_mock, client_mock):