diff --git a/changelogs/fragments/488-ecs_service-support_exec.yml b/changelogs/fragments/488-ecs_service-support_exec.yml
new file mode 100644
index 00000000000..70bc3256d4d
--- /dev/null
+++ b/changelogs/fragments/488-ecs_service-support_exec.yml
@@ -0,0 +1,3 @@
+minor_changes:
+- ecs_service - added new parameter ``enable_execute_command`` (https://github.com/ansible-collections/community.aws/pull/488).
+- ecs_service - handle SDK errors more cleanly on update failures (https://github.com/ansible-collections/community.aws/pull/488).
diff --git a/plugins/modules/ecs_service.py b/plugins/modules/ecs_service.py
index 64963d5c695..2d86a6bd588 100644
--- a/plugins/modules/ecs_service.py
+++ b/plugins/modules/ecs_service.py
@@ -132,6 +132,13 @@
               rollback:
                 type: bool
                 description: If enabled, ECS will roll back your service to the last completed deployment after a failure.
+    enable_execute_command:
+        description:
+          - Whether or not to enable the execute command functionality for the containers in the ECS task.
+          - If I(enable_execute_command=true) execute command functionality is enabled on all containers in the ECS task.
+        required: false
+        type: bool
+        version_added: 5.4.0
     placement_constraints:
         description:
           - The placement constraints for the tasks in the service.
@@ -776,6 +783,9 @@ def is_matching_service(self, expected, existing):
         if boto3_tag_list_to_ansible_dict(existing.get('tags', [])) != (expected['tags'] or {}):
             return False
 
+        if (expected["enable_execute_command"] or False) != existing.get("enableExecuteCommand", False):
+            return False
+
         # expected is params. DAEMON scheduling strategy returns desired count equal to
         # number of instances running; don't check desired count if scheduling strat is daemon
         if (expected['scheduling_strategy'] != 'DAEMON'):
@@ -784,11 +794,30 @@ def is_matching_service(self, expected, existing):
 
         return True
 
-    def create_service(self, service_name, cluster_name, task_definition, load_balancers,
-                       desired_count, client_token, role, deployment_controller, deployment_configuration,
-                       placement_constraints, placement_strategy, health_check_grace_period_seconds,
-                       network_configuration, service_registries, launch_type, platform_version,
-                       scheduling_strategy, capacity_provider_strategy, tags, propagate_tags):
+    def create_service(
+        self,
+        service_name,
+        cluster_name,
+        task_definition,
+        load_balancers,
+        desired_count,
+        client_token,
+        role,
+        deployment_controller,
+        deployment_configuration,
+        placement_constraints,
+        placement_strategy,
+        health_check_grace_period_seconds,
+        network_configuration,
+        service_registries,
+        launch_type,
+        platform_version,
+        scheduling_strategy,
+        capacity_provider_strategy,
+        tags,
+        propagate_tags,
+        enable_execute_command,
+    ):
 
         params = dict(
             cluster=cluster_name,
@@ -834,14 +863,30 @@ def create_service(self, service_name, cluster_name, task_definition, load_balan
 
         if scheduling_strategy:
             params['schedulingStrategy'] = scheduling_strategy
+        if enable_execute_command:
+            params["enableExecuteCommand"] = enable_execute_command
+
         response = self.ecs.create_service(**params)
         return self.jsonize(response['service'])
 
-    def update_service(self, service_name, cluster_name, task_definition, desired_count,
-                       deployment_configuration, placement_constraints, placement_strategy,
-                       network_configuration, health_check_grace_period_seconds,
-                       force_new_deployment, capacity_provider_strategy, load_balancers,
-                       purge_placement_constraints, purge_placement_strategy):
+    def update_service(
+        self,
+        service_name,
+        cluster_name,
+        task_definition,
+        desired_count,
+        deployment_configuration,
+        placement_constraints,
+        placement_strategy,
+        network_configuration,
+        health_check_grace_period_seconds,
+        force_new_deployment,
+        capacity_provider_strategy,
+        load_balancers,
+        purge_placement_constraints,
+        purge_placement_strategy,
+        enable_execute_command,
+    ):
         params = dict(
             cluster=cluster_name,
             service=service_name,
@@ -873,11 +918,14 @@ def update_service(self, service_name, cluster_name, task_definition, desired_co
         # desired count is not required if scheduling strategy is daemon
         if desired_count is not None:
             params['desiredCount'] = desired_count
+        if enable_execute_command is not None:
+            params["enableExecuteCommand"] = enable_execute_command
 
         if load_balancers:
             params['loadBalancers'] = load_balancers
 
         response = self.ecs.update_service(**params)
+
         return self.jsonize(response['service'])
 
     def jsonize(self, service):
@@ -965,8 +1013,9 @@ def main():
                 base=dict(type='int')
             )
         ),
-        propagate_tags=dict(required=False, choices=['TASK_DEFINITION', 'SERVICE']),
-        tags=dict(required=False, type='dict'),
+        propagate_tags=dict(required=False, choices=["TASK_DEFINITION", "SERVICE"]),
+        tags=dict(required=False, type="dict"),
+        enable_execute_command=dict(required=False, type="bool"),
     )
 
     module = AnsibleAWSModule(argument_spec=argument_spec,
@@ -1079,47 +1128,54 @@ def main():
                     if task_definition is None and module.params['force_new_deployment']:
                         task_definition = existing['taskDefinition']
 
-                    # update required
-                    response = service_mgr.update_service(module.params['name'],
-                                                          module.params['cluster'],
-                                                          task_definition,
-                                                          module.params['desired_count'],
-                                                          deploymentConfiguration,
-                                                          module.params['placement_constraints'],
-                                                          module.params['placement_strategy'],
-                                                          network_configuration,
-                                                          module.params['health_check_grace_period_seconds'],
-                                                          module.params['force_new_deployment'],
-                                                          capacityProviders,
-                                                          updatedLoadBalancers,
-                                                          module.params['purge_placement_constraints'],
-                                                          module.params['purge_placement_strategy'],
-                                                          )
+                    try:
+                        # update required
+                        response = service_mgr.update_service(
+                            module.params["name"],
+                            module.params["cluster"],
+                            task_definition,
+                            module.params["desired_count"],
+                            deploymentConfiguration,
+                            module.params["placement_constraints"],
+                            module.params["placement_strategy"],
+                            network_configuration,
+                            module.params["health_check_grace_period_seconds"],
+                            module.params["force_new_deployment"],
+                            capacityProviders,
+                            updatedLoadBalancers,
+                            module.params["purge_placement_constraints"],
+                            module.params["purge_placement_strategy"],
+                            module.params["enable_execute_command"],
+                        )
+                    except (botocore.exceptions.BotoCoreError, botocore.exceptions.ClientError) as e:
+                        module.fail_json_aws(e, msg="Couldn't create service")
 
                 else:
                     try:
-                        response = service_mgr.create_service(module.params['name'],
-                                                              module.params['cluster'],
-                                                              module.params['task_definition'],
-                                                              loadBalancers,
-                                                              module.params['desired_count'],
-                                                              clientToken,
-                                                              role,
-                                                              deploymentController,
-                                                              deploymentConfiguration,
-                                                              module.params['placement_constraints'],
-                                                              module.params['placement_strategy'],
-                                                              module.params['health_check_grace_period_seconds'],
-                                                              network_configuration,
-                                                              serviceRegistries,
-                                                              module.params['launch_type'],
-                                                              module.params['platform_version'],
-                                                              module.params['scheduling_strategy'],
-                                                              capacityProviders,
-                                                              module.params['tags'],
-                                                              module.params['propagate_tags'],
-                                                              )
-                    except botocore.exceptions.ClientError as e:
+                        response = service_mgr.create_service(
+                            module.params["name"],
+                            module.params["cluster"],
+                            module.params["task_definition"],
+                            loadBalancers,
+                            module.params["desired_count"],
+                            clientToken,
+                            role,
+                            deploymentController,
+                            deploymentConfiguration,
+                            module.params["placement_constraints"],
+                            module.params["placement_strategy"],
+                            module.params["health_check_grace_period_seconds"],
+                            network_configuration,
+                            serviceRegistries,
+                            module.params["launch_type"],
+                            module.params["platform_version"],
+                            module.params["scheduling_strategy"],
+                            capacityProviders,
+                            module.params["tags"],
+                            module.params["propagate_tags"],
+                            module.params["enable_execute_command"],
+                        )
+                    except (botocore.exceptions.BotoCoreError, botocore.exceptions.ClientError) as e:
                         module.fail_json_aws(e, msg="Couldn't create service")
 
                 if response.get('tags', None):
diff --git a/tests/integration/targets/ecs_cluster/tasks/01_create_requirements.yml b/tests/integration/targets/ecs_cluster/tasks/01_create_requirements.yml
index ea2709a617d..31ca3cf275e 100644
--- a/tests/integration/targets/ecs_cluster/tasks/01_create_requirements.yml
+++ b/tests/integration/targets/ecs_cluster/tasks/01_create_requirements.yml
@@ -7,6 +7,19 @@
     managed_policy:
     - AmazonEC2ContainerServiceRole
     wait: True
+  register: iam_role_creation
+
+- name: ensure AmazonECSTaskExecutionRolePolicy exists
+  iam_role:
+    name: "{{ ecs_task_role_name }}"
+    assume_role_policy_document: "{{ lookup('file','ecs-trust-policy.json') }}"
+    description: "Allows ECS containers to make calls to ECR"
+    state: present
+    create_instance_profile: false
+    managed_policy:
+      - AmazonECSTaskExecutionRolePolicy
+    wait: True
+  register: iam_execution_role
 
 - name: ensure AWSServiceRoleForECS role exists
   iam_role_info:
diff --git a/tests/integration/targets/ecs_cluster/tasks/20_ecs_service.yml b/tests/integration/targets/ecs_cluster/tasks/20_ecs_service.yml
index 77d8901c430..2e44f5bbcae 100644
--- a/tests/integration/targets/ecs_cluster/tasks/20_ecs_service.yml
+++ b/tests/integration/targets/ecs_cluster/tasks/20_ecs_service.yml
@@ -214,6 +214,44 @@
       - not ecs_service_scale_down.changed
       - ecs_service_scale_down.service.desiredCount == 0
 
+- name: update task definition
+  ecs_taskdefinition:
+    containers: "{{ ecs_task_containers }}"
+    family: "{{ ecs_task_name }}"
+    task_role_arn: "{{ ecs_task_role_name }}"
+    state: present
+  register: ecs_task_update
+
+- name: check that initial task definition changes
+  assert:
+    that:
+      - ecs_task_update.changed
+
+- name: Enable ExecuteCommand
+  ecs_service:
+    state: present
+    name: "{{ ecs_service_name }}"
+    cluster: "{{ ecs_cluster_name }}"
+    task_definition: "{{ ecs_task_name }}:{{ ecs_task_update.taskdefinition.revision }}"
+    desired_count: 0
+    deployment_configuration: "{{ ecs_service_deployment_configuration }}"
+    placement_strategy: "{{ ecs_service_placement_strategy }}"
+    placement_constraints:
+      - type: distinctInstance
+    health_check_grace_period_seconds: "{{ ecs_service_health_check_grace_period }}"
+    load_balancers:
+      - targetGroupArn: "{{ elb_target_group_instance.target_group_arn }}"
+        containerName: "{{ ecs_task_name }}"
+        containerPort: "{{ ecs_task_container_port }}"
+    role: "{{ ecs_service_role_name }}"
+    enable_execute_command: True
+  register: ecs_service_execute
+
+- name: check that ECS service changed
+  assert:
+    that:
+      - ecs_service_execute.changed
+
 - name: delete ECS service definition
   ecs_service:
     state: absent
@@ -682,25 +720,10 @@
     that:
       - ecs_service_remove_strategy.changed
       - "ecs_service_remove_strategy.service.placementStrategy | length == 0"
+
 # ============================================================
 # Begin tests for Fargate
 
-- name: ensure AmazonECSTaskExecutionRolePolicy exists
-  iam_role:
-    name: "{{ ecs_task_role_name }}"
-    assume_role_policy_document: "{{ lookup('file','ecs-trust-policy.json') }}"
-    description: "Allows ECS containers to make calls to ECR"
-    state: present
-    create_instance_profile: false
-    managed_policy:
-      - AmazonECSTaskExecutionRolePolicy
-    wait: True
-  register: iam_execution_role
-
-- name: pause for iam availability
-  ansible.builtin.pause:
-    seconds: 20
-
 - name: create Fargate VPC-networked task definition with host port set to 8080 and unsupported network mode (expected to fail)
   ecs_taskdefinition:
     containers: "{{ ecs_fargate_task_containers }}"