diff --git a/changelogs/fragments/853-aws_ssm-sudo.yml b/changelogs/fragments/853-aws_ssm-sudo.yml
new file mode 100644
index 00000000000..fbbd423c02f
--- /dev/null
+++ b/changelogs/fragments/853-aws_ssm-sudo.yml
@@ -0,0 +1,9 @@
+breaking_changes:
+- aws_ssm - the AWS SSM plugin was incorrectly prepending ``sudo`` to most commands.  This behaviour was incorrect and has been removed.
+  To execute commands as a specific user, including the ``root`` user, the ``become`` and ``become_user`` directives should be used.  See the
+  `Ansible documentation for more information <https://docs.ansible.com/ansible/latest/playbook_guide/playbooks_privilege_escalation.html>`_
+  (https://github.com/ansible-collections/community.aws/issues/853).
+
+minor_changes:
+- aws_ssm - Updated the documentation to explicitly mention that the ``ansible_user`` and ``remote_user`` variables are not supported by the
+  plugin (https://github.com/ansible-collections/community.aws/pull/1682).
diff --git a/plugins/connection/aws_ssm.py b/plugins/connection/aws_ssm.py
index 0b1e12ff94a..b6c246c2c46 100644
--- a/plugins/connection/aws_ssm.py
+++ b/plugins/connection/aws_ssm.py
@@ -7,18 +7,27 @@
 __metaclass__ = type
 
 DOCUMENTATION = '''
-author:
-- Pat Sharkey (@psharkey) <psharkey@cleo.com>
-- HanumanthaRao MVL (@hanumantharaomvl) <hanumanth@flux7.com>
-- Gaurav Ashtikar (@gau1991) <gaurav.ashtikar@flux7.com>
 name: aws_ssm
-short_description: execute via AWS Systems Manager
+author:
+  - Pat Sharkey (@psharkey) <psharkey@cleo.com>
+  - HanumanthaRao MVL (@hanumantharaomvl) <hanumanth@flux7.com>
+  - Gaurav Ashtikar (@gau1991) <gaurav.ashtikar@flux7.com>
+
+short_description: connect to EC2 instances via AWS Systems Manager
 description:
-- This connection plugin allows ansible to execute tasks on an EC2 instance via the aws ssm CLI.
+  - This connection plugin allows Ansible to execute tasks on an EC2 instance via an AWS SSM Session.
+notes:
+  - The M(community.aws.aws_ssm) connection plugin does not support using the ``remote_user`` and
+    ``ansible_user`` variables to configure the remote user.  The ``become_user`` parameter should
+    be used to configure which user to run commands as.  Remote commands will often default to
+    running as the ``ssm-agent`` user, however this will also depend on how SSM has been configured.
 requirements:
-- The remote EC2 instance must be running the AWS Systems Manager Agent (SSM Agent).
-- The control machine must have the aws session manager plugin installed.
-- The remote EC2 linux instance must have the curl installed.
+  - The remote EC2 instance must be running the AWS Systems Manager Agent (SSM Agent).
+    U(https://docs.aws.amazon.com/systems-manager/latest/userguide/session-manager-getting-started.html)
+  - The control machine must have the AWS session manager plugin installed.
+    U(https://docs.aws.amazon.com/systems-manager/latest/userguide/session-manager-working-with-install-plugin.html)
+  - The remote EC2 Linux instance must have curl installed.
+
 options:
   access_key_id:
     description: The STS access key to use when connecting via session-manager.
@@ -83,7 +92,11 @@
     vars:
     - name: ansible_aws_ssm_bucket_sse_kms_key_id
   ssm_document:
-    description: SSM document to use when connecting.
+    description:
+    - SSM Session document to use when connecting.
+    - To configure the remote_user (when C(become=False), it is possible to use an SSM Session
+      document and define the C(runAsEnabled) and C(runAsDefaultUser) parameters.  See also
+      U(https://docs.aws.amazon.com/systems-manager/latest/userguide/session-manager-schema.html)
     vars:
     - name: ansible_aws_ssm_document
     version_added: 5.2.0
@@ -318,11 +331,14 @@ def chunks(lst, n):
 class Connection(ConnectionBase):
     ''' AWS SSM based connections '''
 
-    transport = 'community.aws.aws_ssm'
+    transport = "community.aws.aws_ssm"
+    default_user = ""
+
     allow_executable = False
     allow_extras = True
     has_pipelining = False
     is_windows = False
+
     _client = None
     _s3_client = None
     _session = None
@@ -430,6 +446,7 @@ def _connect(self):
     def reset(self):
         ''' start a fresh ssm session '''
         self._vvvv('reset called on ssm connection')
+        self.close()
         return self.start_session()
 
     def start_session(self):
@@ -646,8 +663,6 @@ def _wrap_command(self, cmd, sudoable, mark_start, mark_end):
                 cmd = self._shell._encode_script(cmd, preserve_rc=True)
             cmd = cmd + "; echo " + mark_start + "\necho " + mark_end + "\n"
         else:
-            if sudoable:
-                cmd = "sudo " + cmd
             cmd = (
                 f"printf '%s\\n' '{mark_start}';\n"
                 f"echo | {cmd};\n"
diff --git a/tests/integration/targets/connection/test_assume.yml b/tests/integration/targets/connection/test_assume.yml
new file mode 100644
index 00000000000..e8c6aab0a09
--- /dev/null
+++ b/tests/integration/targets/connection/test_assume.yml
@@ -0,0 +1,16 @@
+- name: 'Ensure remote user exists'
+  ansible.builtin.user:
+    name: '{{ user_name }}'
+    shell: /bin/bash
+  become_user: 'root'
+  become: True
+
+- name: 'Attempt to run a shell command as the user ({{ user_name }})'
+  become_user: '{{ user_name }}'
+  become: True
+  command: 'id -u -n'
+  register: id_cmd
+
+- assert:
+    that:
+      - id_cmd.stdout == '{{ user_name }}'
diff --git a/tests/integration/targets/connection/test_connection.yml b/tests/integration/targets/connection/test_connection.yml
index 87ed5851dbb..4919898ebb6 100644
--- a/tests/integration/targets/connection/test_connection.yml
+++ b/tests/integration/targets/connection/test_connection.yml
@@ -9,9 +9,12 @@
   tasks:
 
   ### test wait_for_connection plugin
+
   - wait_for_connection:
       timeout: '{{ wait_for_timeout | default(100) }}'
 
+  ### Try to gather the default facts from the host
+
   - name: Gather facts
     ansible.builtin.setup:
 
@@ -50,3 +53,27 @@
     local_action: file path={{ local_file }} state=absent
   - name: remove remote temp file
     action: "{{ action_prefix }}file path={{ remote_file }} state=absent"
+
+  ### Test that we're the user we expect to be and can change where appropriate
+  # Regression - https://github.com/ansible-collections/community.aws/issues/853
+
+  - name: Test user manipulaton
+    when:
+      - '"aws_ssm_linux" in groups'
+    block:
+      - name: 'Find ID when become=False'
+        become: False
+        command: 'id -u -n'
+        register: id_cmd
+
+      - assert:
+          that:
+            - id_cmd.stdout == 'ssm-user'
+
+      - include_tasks: 'test_assume.yml'
+        loop:
+          - ssm-agent
+          - zuul
+          - root
+        loop_control:
+          loop_var: user_name
diff --git a/tests/integration/targets/setup_connection_aws_ssm/tasks/cleanup.yml b/tests/integration/targets/setup_connection_aws_ssm/tasks/cleanup.yml
index 6171e5eb630..15b833213ba 100644
--- a/tests/integration/targets/setup_connection_aws_ssm/tasks/cleanup.yml
+++ b/tests/integration/targets/setup_connection_aws_ssm/tasks/cleanup.yml
@@ -68,6 +68,7 @@
       iam_role:
         name: "{{ iam_role_name }}"
         state: absent
+        delete_instance_profile: True
       ignore_errors: yes
       when: iam_role_vars_file.stat.exists == true