ecs_taskdefinition: use aws_retry to avoid throttling exception (#2124)

SUMMARY

Fixes #2123 by adding aws_retry=True to the API calls.

ISSUE TYPE

Bugfix Pull Request

COMPONENT NAME

ecs_taskdefinition
ADDITIONAL INFORMATION

We observed that ecs_taskdefinition intermittently causes a ThrottlingException when running on a task definition with a large number of revisions. Looking at the code, it appears that describe_task_definitions loops over the revisions without using the retry mechanism. This PR attempts to solve the problem by adding aws_retry=True to the API calls.
Due to the nature of the problem (intermittent throttling by AWS), I couldn't devise automated tests that validate the fix.

Reviewed-by: Alina Buzachis
Reviewed-by: Mark Chappell
Reviewed-by: Eli Acherkan
(cherry picked from commit 97131ec)

Author: eacherkan-aternity

changelogs/fragments/2124-add-retry-to-ecs_taskdefinition.yml

@@ -0,0 +1,2 @@
+bugfixes:
+  - ecs_taskdefinition - avoid throttling exceptions on task definitions with a large number of revisions by using the retry mechanism (https://github.com/ansible-collections/community.aws/issues/2123).

plugins/modules/ecs_taskdefinition.py

@@ -916,7 +916,7 @@ def fetch():
             if data["nextToken"]:
                 params["nextToken"] = data["nextToken"]
 
-            result = self.ecs.list_task_definitions(**params)
+            result = self.ecs.list_task_definitions(aws_retry=True, **params)
             data["taskDefinitionArns"] += result["taskDefinitionArns"]
             data["nextToken"] = result.get("nextToken", None)
             return data["nextToken"] is not None
@@ -929,15 +929,15 @@ def fetch():
         return list(
             sorted(
                 [
-                    self.ecs.describe_task_definition(taskDefinition=arn)["taskDefinition"]
+                    self.ecs.describe_task_definition(aws_retry=True, taskDefinition=arn)["taskDefinition"]
                     for arn in data["taskDefinitionArns"]
                 ],
                 key=lambda td: td["revision"],
             )
         )
 
     def deregister_task(self, taskArn):
-        response = self.ecs.deregister_task_definition(taskDefinition=taskArn)
+        response = self.ecs.deregister_task_definition(aws_retry=True, taskDefinition=taskArn)
         return response["taskDefinition"]