diff --git a/changelogs/fragments/1209-ecs_service-add-waiters.yml b/changelogs/fragments/1209-ecs_service-add-waiters.yml
new file mode 100644
index 00000000000..cfe3df4af58
--- /dev/null
+++ b/changelogs/fragments/1209-ecs_service-add-waiters.yml
@@ -0,0 +1,3 @@
+minor_changes:
+  - ecs_service - add ``wait`` parameter and waiter for deleting services (https://github.com/ansible-collections/community.aws/pull/1209).
+  - ecs_task - add ``wait`` parameter and waiter for running and stopping tasks (https://github.com/ansible-collections/community.aws/pull/1209).
diff --git a/plugins/modules/ecs_service.py b/plugins/modules/ecs_service.py
index f7bd5779e18..52b7bcd9a56 100644
--- a/plugins/modules/ecs_service.py
+++ b/plugins/modules/ecs_service.py
@@ -218,6 +218,13 @@
         required: false
         choices: ["DAEMON", "REPLICA"]
         type: str
+    wait:
+        description:
+          - Whether or not to wait for the service to be inactive.
+          - Waits only when I(state) is C(absent).
+        type: bool
+        default: false
+        version_added: 4.1.0
 extends_documentation_fragment:
 - amazon.aws.aws
 - amazon.aws.ec2
@@ -728,6 +735,7 @@ def main():
         force_new_deployment=dict(required=False, default=False, type='bool'),
         force_deletion=dict(required=False, default=False, type='bool'),
         deployment_configuration=dict(required=False, default={}, type='dict'),
+        wait=dict(required=False, default=False, type='bool'),
         placement_constraints=dict(
             required=False,
             default=[],
@@ -912,8 +920,24 @@ def main():
                             module.params['cluster'],
                             module.params['force_deletion'],
                         )
+
+                        # Wait for service to be INACTIVE prior to exiting
+                        if module.params['wait']:
+                            waiter = service_mgr.ecs.get_waiter('services_inactive')
+                            try:
+                                waiter.wait(
+                                    services=[module.params['name']],
+                                    cluster=module.params['cluster'],
+                                    WaiterConfig={
+                                        'Delay': module.params['delay'],
+                                        'MaxAttempts': module.params['repeat']
+                                    }
+                                )
+                            except botocore.exceptions.WaiterError as e:
+                                module.fail_json_aws(e, 'Timeout waiting for service removal')
                     except botocore.exceptions.ClientError as e:
                         module.fail_json_aws(e, msg="Couldn't delete service")
+
                 results['changed'] = True
 
     elif module.params['state'] == 'deleting':
diff --git a/plugins/modules/ecs_task.py b/plugins/modules/ecs_task.py
index b4c625df712..893bf380ba6 100644
--- a/plugins/modules/ecs_task.py
+++ b/plugins/modules/ecs_task.py
@@ -90,6 +90,12 @@
           - Tags that will be added to ecs tasks on start and run
         required: false
         aliases: ['resource_tags']
+    wait:
+        description:
+          - Whether or not to wait for the desired state.
+        type: bool
+        default: false
+        version_added: 4.1.0
 extends_documentation_fragment:
     - amazon.aws.aws
     - amazon.aws.ec2
@@ -351,7 +357,8 @@ def main():
         started_by=dict(required=False, type='str'),  # R S
         network_configuration=dict(required=False, type='dict'),
         launch_type=dict(required=False, choices=['EC2', 'FARGATE']),
-        tags=dict(required=False, type='dict', aliases=['resource_tags'])
+        tags=dict(required=False, type='dict', aliases=['resource_tags']),
+        wait=dict(required=False, default=False, type='bool'),
     )
 
     module = AnsibleAWSModule(argument_spec=argument_spec, supports_check_mode=True,
@@ -393,7 +400,9 @@ def main():
             results['task'] = existing
         else:
             if not module.check_mode:
-                results['task'] = service_mgr.run_task(
+
+                # run_task returns a list of tasks created
+                tasks = service_mgr.run_task(
                     module.params['cluster'],
                     module.params['task_definition'],
                     module.params['overrides'],
@@ -402,6 +411,21 @@ def main():
                     module.params['launch_type'],
                     module.params['tags'],
                 )
+
+                # Wait for task(s) to be running prior to exiting
+                if module.params['wait']:
+
+                    waiter = service_mgr.ecs.get_waiter('tasks_running')
+                    try:
+                        waiter.wait(
+                            tasks=[task['taskArn'] for task in tasks],
+                            cluster=module.params['cluster'],
+                        )
+                    except botocore.exceptions.WaiterError as e:
+                        module.fail_json_aws(e, 'Timeout waiting for tasks to run')
+
+                results['task'] = tasks
+
             results['changed'] = True
 
     elif module.params['operation'] == 'start':
@@ -418,6 +442,7 @@ def main():
                     module.params['started_by'],
                     module.params['tags'],
                 )
+
             results['changed'] = True
 
     elif module.params['operation'] == 'stop':
@@ -431,6 +456,19 @@ def main():
                     module.params['cluster'],
                     module.params['task']
                 )
+
+                # Wait for task to be stopped prior to exiting
+                if module.params['wait']:
+
+                    waiter = service_mgr.ecs.get_waiter('tasks_stopped')
+                    try:
+                        waiter.wait(
+                            tasks=[module.params['task']],
+                            cluster=module.params['cluster'],
+                        )
+                    except botocore.exceptions.WaiterError as e:
+                        module.fail_json_aws(e, 'Timeout waiting for task to stop')
+
             results['changed'] = True
 
     module.exit_json(**results)
diff --git a/tests/integration/targets/ecs_cluster/tasks/main.yml b/tests/integration/targets/ecs_cluster/tasks/main.yml
index 7f49374b532..7f3a1deee85 100644
--- a/tests/integration/targets/ecs_cluster/tasks/main.yml
+++ b/tests/integration/targets/ecs_cluster/tasks/main.yml
@@ -326,15 +326,40 @@
         role: "ecsServiceRole"
       register: ecs_service_scale_down
 
-    - name: pause to allow service to scale down
-      pause:
-        seconds: 60
+    - name: assert that ECS service is scaled down
+      assert:
+        that:
+          - ecs_service_scale_down.changed
+          - ecs_service_scale_down.service.desiredCount == 0
+
+    - name: scale down ECS service again
+      ecs_service:
+        state: present
+        name: "{{ ecs_service_name }}"
+        cluster: "{{ ecs_cluster_name }}"
+        task_definition: "{{ ecs_task_name }}:{{ ecs_task_definition.taskdefinition.revision }}"
+        desired_count: 0
+        deployment_configuration: "{{ ecs_service_deployment_configuration }}"
+        placement_strategy: "{{ ecs_service_placement_strategy }}"
+        load_balancers:
+          - targetGroupArn: "{{ elb_target_group_instance.target_group_arn }}"
+            containerName: "{{ ecs_task_name }}"
+            containerPort: "{{ ecs_task_container_port }}"
+        role: "ecsServiceRole"
+      register: ecs_service_scale_down
+
+    - name: assert no change
+      assert:
+        that:
+          - not ecs_service_scale_down.changed
+          - ecs_service_scale_down.service.desiredCount == 0
 
     - name: delete ECS service definition
       ecs_service:
         state: absent
         name: "{{ ecs_service_name }}"
         cluster: "{{ ecs_cluster_name }}"
+        wait: yes
       register: delete_ecs_service
 
     - name: assert that deleting ECS service worked
@@ -342,10 +367,17 @@
         that:
           - delete_ecs_service.changed
 
-    - name: assert that deleting ECS service worked
+    - name: delete ECS service definition again
+      ecs_service:
+        state: absent
+        name: "{{ ecs_service_name }}"
+        cluster: "{{ ecs_cluster_name }}"
+      register: delete_ecs_service
+
+    - name: assert no change
       assert:
         that:
-          - delete_ecs_service.changed
+          - not delete_ecs_service.changed
 
     - name: create VPC-networked task definition with host port set to 0 (expected to fail)
       ecs_taskdefinition:
@@ -382,10 +414,6 @@
         that:
           - "ecs_taskdefinition_info.network_mode == 'awsvpc'"
 
-    - name: pause to allow service to scale down
-      pause:
-        seconds: 60
-
     - name: create ECS service definition with network configuration
       ecs_service:
         state: present
@@ -428,7 +456,6 @@
         state: present
       register: ecs_service_creation_hcgp
 
-
     - name: health_check_grace_period_seconds sets HealthChecGracePeriodSeconds
       assert:
         that:
@@ -525,7 +552,7 @@
     - name: attempt to get facts from missing task definition
       ecs_taskdefinition_info:
         task_definition: "{{ ecs_task_name }}-vpc:{{ ecs_task_definition.taskdefinition.revision + 1}}"
-    
+
     - name: Create another task definition with placement constraints
       ecs_taskdefinition:
         containers: "{{ ecs_task_containers }}"
@@ -540,19 +567,31 @@
           - ecs_task_definition_constraints is changed
           - ecs_task_definition_constraints.taskdefinition.placementConstraints[0].type == "{{ ecs_taskdefinition_placement_constraints[0].type }}"
           - ecs_task_definition_constraints.taskdefinition.placementConstraints[0].expression == "{{ ecs_taskdefinition_placement_constraints[0].expression }}"
-    
+
     - name: Remove ecs task definition with placement constraints
       ecs_taskdefinition:
         containers: "{{ ecs_task_containers }}"
         arn: "{{ ecs_task_definition_constraints.taskdefinition.taskDefinitionArn }}"
         state: absent
       register: ecs_task_definition_constraints_delete
-  
+
     - name: Check that task definition has been deleted
       assert:
         that:
           - ecs_task_definition_constraints_delete is changed
 
+    - name: Remove ecs task definition with placement constraints again
+      ecs_taskdefinition:
+        containers: "{{ ecs_task_containers }}"
+        arn: "{{ ecs_task_definition_constraints.taskdefinition.taskDefinitionArn }}"
+        state: absent
+      register: ecs_task_definition_constraints_delete
+
+    - name: Assert no change
+      assert:
+        that:
+          - ecs_task_definition_constraints_delete is not changed
+
     # ============================================================
     # Begin tests for Fargate
 
@@ -674,6 +713,8 @@
         that:
           - 'ecs_fargate_service_network_with_awsvpc.service.networkConfiguration.awsvpcConfiguration.assignPublicIp == "ENABLED"'
 
+    ### FIX - run tasks are all failing with CannotPullContainerError in AWS
+    ### So using wait: True fails when waiting for tasks to be started
     - name: create fargate ECS task with run task
       ecs_task:
         operation: run
@@ -687,8 +728,35 @@
             - '{{ setup_sg.group_id }}'
           assign_public_ip: true
         started_by: ansible_user
+        # wait: yes
       register: fargate_run_task_output
 
+    - name: Assert changed
+      assert:
+        that:
+          - fargate_run_task_output.changed
+
+    # - name: create fargate ECS task with run task again
+    #   ecs_task:
+    #     operation: run
+    #     cluster: "{{ ecs_cluster_name }}"
+    #     task_definition: "{{ ecs_task_name }}-vpc"
+    #     launch_type: FARGATE
+    #     count: 1
+    #     network_configuration:
+    #       subnets: "{{ setup_subnet.results | map(attribute='subnet.id') | list }}"
+    #       security_groups:
+    #         - '{{ setup_sg.group_id }}'
+    #       assign_public_ip: true
+    #     started_by: ansible_user
+    #   register: fargate_run_task_output
+
+    # - name: Assert no change
+    #   assert:
+    #     that:
+    #       - not fargate_run_task_output.changed
+
+    ### This does not fail
     - name: create fargate ECS task with run task and tags (LF disabled) (should fail)
       ecs_task:
         operation: run
@@ -708,6 +776,11 @@
       register: fargate_run_task_output_with_tags_fail
       ignore_errors: yes
 
+    # - name: assert that using Fargate ECS service fails
+    #   assert:
+    #     that:
+    #       - fargate_run_task_output_with_tags_fail is failed
+
     - name: enable taskLongArnFormat
       command: aws ecs put-account-setting --name taskLongArnFormat --value enabled
       environment:
@@ -865,26 +938,19 @@
       ignore_errors: yes
       register: ecs_service_scale_down
 
-    - name: stop Fargate ECS task
-      ecs_task:
-        task: "{{ fargate_run_task_output.task[0].taskArn }}"
-        task_definition: "{{ ecs_task_name }}-vpc"
-        operation: stop
-        cluster: "{{ ecs_cluster_name }}"
-      ignore_errors: yes
-
-    - name: stop Fargate ECS task
+    - name: stop Fargate ECS tasks
       ecs_task:
-        task: "{{ fargate_run_task_output_with_tags.task[0].taskArn }}"
+        task: "{{ item.task[0].taskArn }}"
         task_definition: "{{ ecs_task_name }}-vpc"
         operation: stop
         cluster: "{{ ecs_cluster_name }}"
+        wait: yes
       ignore_errors: yes
-
-    - name: pause to allow services to scale down
-      pause:
-        seconds: 60
-      when: ecs_service_scale_down is not failed
+      with_items:
+        - "{{ fargate_run_task_output }}"
+        - "{{ fargate_run_task_output_with_tags }}"
+        - "{{ fargate_run_task_output_with_assign_ip }}"
+        - "{{ fargate_run_task_output_with_tags_fail }}"
 
     - name: remove ecs service
       ecs_service:
@@ -892,6 +958,7 @@
         cluster: "{{ ecs_cluster_name }}"
         name: "{{ ecs_service_name }}"
         force_deletion: yes
+        wait: yes
       ignore_errors: yes
 
     - name: remove second ecs service
@@ -900,6 +967,7 @@
         cluster: "{{ ecs_cluster_name }}"
         name: "{{ ecs_service_name }}2"
         force_deletion: yes
+        wait: yes
       ignore_errors: yes
 
     - name: remove mft ecs service
@@ -908,6 +976,7 @@
         cluster: "{{ ecs_cluster_name }}"
         name: "{{ ecs_service_name }}-mft"
         force_deletion: yes
+        wait: yes
       ignore_errors: yes
 
     - name: remove scheduling_strategy ecs service
@@ -916,6 +985,7 @@
         cluster: "{{ ecs_cluster_name }}"
         name: "{{ ecs_service_name }}-replica"
         force_deletion: yes
+        wait: yes
       ignore_errors: yes
 
     - name: remove fargate ECS service
@@ -924,6 +994,7 @@
         name: "{{ ecs_service_name }}4"
         cluster: "{{ ecs_cluster_name }}"
         force_deletion: yes
+        wait: yes
       ignore_errors: yes
       register: ecs_fargate_service_network_with_awsvpc
 
@@ -965,6 +1036,14 @@
         state: absent
       ignore_errors: yes
 
+    - name: remove ec2 ecs task definition
+      ecs_taskdefinition:
+        containers: "{{ ecs_fargate_task_containers }}"
+        family: "{{ ecs_task_name }}-vpc"
+        revision: "{{ ecs_ec2_task_definition.taskdefinition.revision }}"
+        state: absent
+      ignore_errors: yes
+
     - name: remove ecs task definition for absent with arn
       ecs_taskdefinition:
         containers: "{{ ecs_task_containers }}"
@@ -981,11 +1060,6 @@
       ignore_errors: yes
       register: elb_application_lb_remove
 
-    - name: pause to allow target group to be disassociated
-      pause:
-        seconds: 30
-      when: not elb_application_lb_remove is failed
-
     - name: remove setup keypair
       ec2_key:
         name: '{{ resource_prefix }}_ecs_cluster'
@@ -998,9 +1072,6 @@
         state: absent
       ignore_errors: yes
       register: this_deletion
-      retries: 12
-      delay: 10
-      until: this_deletion is not failed
 
     - name: remove security groups
       ec2_group: