diff --git a/changelogs/fragments/365-mysql_user-add-on_new_username-and-password_changed.yml b/changelogs/fragments/365-mysql_user-add-on_new_username-and-password_changed.yml
new file mode 100644
index 0000000..2796776
--- /dev/null
+++ b/changelogs/fragments/365-mysql_user-add-on_new_username-and-password_changed.yml
@@ -0,0 +1,10 @@
+minor_changes:
+  - >
+    mysql_user - Add the option ``on_new_username`` to argument ``update_password`` to reuse the password (plugin and
+    authentication_string) when creating a new user if some user with the same name already exists.
+    If the existing user with the same name have varying passwords, the password from the arguments is used like with
+    ``update_password: always`` (https://github.com/ansible-collections/community.mysql/pull/365).
+  - >
+    mysql_user - Add the result field ``password_changed`` (boolean). It is true, when the user got a new password.
+    When the user was created with ``update_password: on_new_username`` and an existing password was reused,
+    ``password_changed`` is false (https://github.com/ansible-collections/community.mysql/pull/365).
diff --git a/plugins/module_utils/user.py b/plugins/module_utils/user.py
index dd0509b..655d847 100644
--- a/plugins/module_utils/user.py
+++ b/plugins/module_utils/user.py
@@ -112,21 +112,49 @@ def get_grants(cursor, user, host):
     return grants.split(", ")
 
 
+def get_existing_authentication(cursor, user):
+    # Return the plugin and auth_string if there is exactly one distinct existing plugin and auth_string.
+    cursor.execute("SELECT VERSION()")
+    if 'mariadb' in cursor.fetchone()[0].lower():
+        # before MariaDB 10.2.19 and 10.3.11, "password" and "authentication_string" can differ
+        # when using mysql_native_password
+        cursor.execute("""select plugin, auth from (
+            select plugin, password as auth from mysql.user where user=%(user)s
+            union select plugin, authentication_string as auth from mysql.user where user=%(user)s
+            ) x group by plugin, auth limit 2
+        """, {'user': user})
+    else:
+        cursor.execute("""select plugin, authentication_string as auth from mysql.user where user=%(user)s
+            group by plugin, authentication_string limit 2""", {'user': user})
+    rows = cursor.fetchall()
+    if len(rows) == 1:
+        return {'plugin': rows[0][0], 'auth_string': rows[0][1]}
+    return None
+
+
 def user_add(cursor, user, host, host_all, password, encrypted,
              plugin, plugin_hash_string, plugin_auth_string, new_priv,
-             tls_requires, check_mode):
+             tls_requires, check_mode, reuse_existing_password):
     # we cannot create users without a proper hostname
     if host_all:
-        return False
+        return {'changed': False, 'password_changed': False}
 
     if check_mode:
-        return True
+        return {'changed': True, 'password_changed': None}
 
     # Determine what user management method server uses
     old_user_mgmt = impl.use_old_user_mgmt(cursor)
 
     mogrify = do_not_mogrify_requires if old_user_mgmt else mogrify_requires
 
+    used_existing_password = False
+    if reuse_existing_password:
+        existing_auth = get_existing_authentication(cursor, user)
+        if existing_auth:
+            plugin = existing_auth['plugin']
+            plugin_hash_string = existing_auth['auth_string']
+            password = None
+            used_existing_password = True
     if password and encrypted:
         if impl.supports_identified_by_password(cursor):
             query_with_args = "CREATE USER %s@%s IDENTIFIED BY PASSWORD %s", (user, host, password)
@@ -156,7 +184,7 @@ def user_add(cursor, user, host, host_all, password, encrypted,
             privileges_grant(cursor, user, host, db_table, priv, tls_requires)
     if tls_requires is not None:
         privileges_grant(cursor, user, host, "*.*", get_grants(cursor, user, host), tls_requires)
-    return True
+    return {'changed': True, 'password_changed': not used_existing_password}
 
 
 def is_hash(password):
@@ -182,6 +210,7 @@ def user_mod(cursor, user, host, host_all, password, encrypted,
     else:
         hostnames = [host]
 
+    password_changed = False
     for host in hostnames:
         # Handle clear text and hashed passwords.
         if not role:
@@ -226,9 +255,10 @@ def user_mod(cursor, user, host, host_all, password, encrypted,
                     encrypted_password = cursor.fetchone()[0]
 
                 if current_pass_hash != encrypted_password:
+                    password_changed = True
                     msg = "Password updated"
                     if module.check_mode:
-                        return (True, msg)
+                        return {'changed': True, 'msg': msg, 'password_changed': password_changed}
                     if old_user_mgmt:
                         cursor.execute("SET PASSWORD FOR %s@%s = %s", (user, host, encrypted_password))
                         msg = "Password updated (old style)"
@@ -280,6 +310,7 @@ def user_mod(cursor, user, host, host_all, password, encrypted,
                     query_with_args = "ALTER USER %s@%s IDENTIFIED WITH %s", (user, host, plugin)
 
                 cursor.execute(*query_with_args)
+                password_changed = True
                 changed = True
 
         # Handle privileges
@@ -297,7 +328,7 @@ def user_mod(cursor, user, host, host_all, password, encrypted,
                         if user != "root" and "PROXY" not in priv:
                             msg = "Privileges updated"
                             if module.check_mode:
-                                return (True, msg)
+                                return {'changed': True, 'msg': msg, 'password_changed': password_changed}
                             privileges_revoke(cursor, user, host, db_table, priv, grant_option, maria_role)
                             changed = True
 
@@ -308,7 +339,7 @@ def user_mod(cursor, user, host, host_all, password, encrypted,
                     if db_table not in curr_priv:
                         msg = "New privileges granted"
                         if module.check_mode:
-                            return (True, msg)
+                            return {'changed': True, 'msg': msg, 'password_changed': password_changed}
                         privileges_grant(cursor, user, host, db_table, priv, tls_requires, maria_role)
                         changed = True
 
@@ -338,7 +369,7 @@ def user_mod(cursor, user, host, host_all, password, encrypted,
                 if len(grant_privs) + len(revoke_privs) > 0:
                     msg = "Privileges updated: granted %s, revoked %s" % (grant_privs, revoke_privs)
                     if module.check_mode:
-                        return (True, msg)
+                        return {'changed': True, 'msg': msg, 'password_changed': password_changed}
                     if len(revoke_privs) > 0:
                         privileges_revoke(cursor, user, host, db_table, revoke_privs, grant_option, maria_role)
                     if len(grant_privs) > 0:
@@ -353,7 +384,7 @@ def user_mod(cursor, user, host, host_all, password, encrypted,
         if current_requires != tls_requires:
             msg = "TLS requires updated"
             if module.check_mode:
-                return (True, msg)
+                return {'changed': True, 'msg': msg, 'password_changed': password_changed}
             if not old_user_mgmt:
                 pre_query = "ALTER USER"
             else:
@@ -369,7 +400,7 @@ def user_mod(cursor, user, host, host_all, password, encrypted,
             cursor.execute(*query_with_args)
             changed = True
 
-    return (changed, msg)
+    return {'changed': changed, 'msg': msg, 'password_changed': password_changed}
 
 
 def user_delete(cursor, user, host, host_all, check_mode):
diff --git a/plugins/modules/mysql_role.py b/plugins/modules/mysql_role.py
index 790c0eb..b37d70d 100644
--- a/plugins/modules/mysql_role.py
+++ b/plugins/modules/mysql_role.py
@@ -911,10 +911,11 @@ class Role():
                                                       set_default_role_all=set_default_role_all)
 
         if privs:
-            changed, msg = user_mod(self.cursor, self.name, self.host,
-                                    None, None, None, None, None, None,
-                                    privs, append_privs, subtract_privs, None,
-                                    self.module, role=True, maria_role=self.is_mariadb)
+            result = user_mod(self.cursor, self.name, self.host,
+                              None, None, None, None, None, None,
+                              privs, append_privs, subtract_privs, None,
+                              self.module, role=True, maria_role=self.is_mariadb)
+            changed = result['changed']
 
         if admin:
             self.role_impl.set_admin(admin)
diff --git a/plugins/modules/mysql_user.py b/plugins/modules/mysql_user.py
index 292179a..c85a910 100644
--- a/plugins/modules/mysql_user.py
+++ b/plugins/modules/mysql_user.py
@@ -118,8 +118,12 @@ options:
     description:
       - C(always) will update passwords if they differ. This affects I(password) and the combination of I(plugin), I(plugin_hash_string), I(plugin_auth_string).
       - C(on_create) will only set the password or the combination of plugin, plugin_hash_string, plugin_auth_string for newly created users.
+      - "C(on_new_username) works like C(on_create), but it tries to reuse an existing password: If one different user
+        with the same username exists, or multiple different users with the same username and equal C(plugin) and
+        C(authentication_string) attribute, the existing C(plugin) and C(authentication_string) are used for the
+        new user instead of the I(password), I(plugin), I(plugin_hash_string) or I(plugin_auth_string) argument."
     type: str
-    choices: [ always, on_create ]
+    choices: [ always, on_create, on_new_username ]
     default: always
   plugin:
     description:
@@ -370,7 +374,7 @@ def main():
         append_privs=dict(type='bool', default=False),
         subtract_privs=dict(type='bool', default=False),
         check_implicit_admin=dict(type='bool', default=False),
-        update_password=dict(type='str', default='always', choices=['always', 'on_create'], no_log=False),
+        update_password=dict(type='str', default='always', choices=['always', 'on_create', 'on_new_username'], no_log=False),
         sql_log_bin=dict(type='bool', default=True),
         plugin=dict(default=None, type='str'),
         plugin_hash_string=dict(default=None, type='str'),
@@ -447,18 +451,22 @@ def main():
         except Exception as e:
             module.fail_json(msg=to_native(e))
         priv = privileges_unpack(priv, mode, ensure_usage=not subtract_privs)
-
+    password_changed = False
     if state == "present":
         if user_exists(cursor, user, host, host_all):
             try:
                 if update_password == "always":
-                    changed, msg = user_mod(cursor, user, host, host_all, password, encrypted,
-                                            plugin, plugin_hash_string, plugin_auth_string,
-                                            priv, append_privs, subtract_privs, tls_requires, module)
+                    result = user_mod(cursor, user, host, host_all, password, encrypted,
+                                      plugin, plugin_hash_string, plugin_auth_string,
+                                      priv, append_privs, subtract_privs, tls_requires, module)
+
                 else:
-                    changed, msg = user_mod(cursor, user, host, host_all, None, encrypted,
-                                            None, None, None,
-                                            priv, append_privs, subtract_privs, tls_requires, module)
+                    result = user_mod(cursor, user, host, host_all, None, encrypted,
+                                      None, None, None,
+                                      priv, append_privs, subtract_privs, tls_requires, module)
+                changed = result['changed']
+                msg = result['msg']
+                password_changed = result['password_changed']
 
             except (SQLParseError, InvalidPrivsError, mysql_driver.Error) as e:
                 module.fail_json(msg=to_native(e))
@@ -468,9 +476,12 @@ def main():
             try:
                 if subtract_privs:
                     priv = None  # avoid granting unwanted privileges
-                changed = user_add(cursor, user, host, host_all, password, encrypted,
-                                   plugin, plugin_hash_string, plugin_auth_string,
-                                   priv, tls_requires, module.check_mode)
+                reuse_existing_password = update_password == 'on_new_username'
+                result = user_add(cursor, user, host, host_all, password, encrypted,
+                                  plugin, plugin_hash_string, plugin_auth_string,
+                                  priv, tls_requires, module.check_mode, reuse_existing_password)
+                changed = result['changed']
+                password_changed = result['password_changed']
                 if changed:
                     msg = "User added"
 
@@ -487,7 +498,7 @@ def main():
         else:
             changed = False
             msg = "User doesn't exist"
-    module.exit_json(changed=changed, user=user, msg=msg)
+    module.exit_json(changed=changed, user=user, msg=msg, password_changed=password_changed)
 
 
 if __name__ == '__main__':
diff --git a/tests/integration/targets/test_mysql_user/tasks/assert_user_password.yml b/tests/integration/targets/test_mysql_user/tasks/assert_user_password.yml
new file mode 100644
index 0000000..fd7e281
--- /dev/null
+++ b/tests/integration/targets/test_mysql_user/tasks/assert_user_password.yml
@@ -0,0 +1,24 @@
+- name: "applying user {{ username }}@{{ host }} with update_password={{ update_password }}"
+  mysql_user:
+    login_user: '{{ mysql_parameters.login_user }}'
+    login_password: '{{ mysql_parameters.login_password }}'
+    login_host: '{{ mysql_parameters.login_host }}'
+    login_port: '{{ mysql_parameters.login_port }}'
+    state: present
+    name: "{{ username }}"
+    host: "{{ host }}"
+    password: "{{ password }}"
+    update_password: "{{ update_password }}"
+  register: result
+- name: assert a change occurred
+  assert:
+    that:
+      - "result.changed == {{ expect_change }}"
+      - "result.password_changed == {{ expect_password_change }}"
+- name: query the user
+  command: "{{ mysql_command }} -BNe \"SELECT plugin, authentication_string FROM mysql.user where user='{{ username }}' and host='{{ host }}'\""
+  register: existing_user
+- name: assert the password is as set to expect_hash
+  assert:
+    that:
+      - "'mysql_native_password\t{{ expect_password_hash }}' in existing_user.stdout_lines"
diff --git a/tests/integration/targets/test_mysql_user/tasks/test_update_password.yml b/tests/integration/targets/test_mysql_user/tasks/test_update_password.yml
new file mode 100644
index 0000000..c9b74bb
--- /dev/null
+++ b/tests/integration/targets/test_mysql_user/tasks/test_update_password.yml
@@ -0,0 +1,128 @@
+# Tests scenarios for both plaintext and encrypted user passwords.
+
+- vars:
+    mysql_parameters:
+      login_user: '{{ mysql_user }}'
+      login_password: '{{ mysql_password }}'
+      login_host: 127.0.0.1
+      login_port: '{{ mysql_primary_port }}'
+    test_password1: kbB9tcx5WOGVGfzV
+    test_password1_hash: '*AF6A7F9D038475C17EE46564F154104877EE5037'
+    test_password2: XBYjpHmjIctMxl1y
+    test_password2_hash: '*9E22D1B35C68BDDF398B8F28AE482E5A865BAC0A'
+    test_password3: tem33JfR5Yx98BB
+    test_password3_hash: '*C7E7C2710702F20336F8D93BC0670C8FB66BDBC7'
+
+
+  block:
+    - include_tasks: assert_user_password.yml
+      vars:
+        username: "{{ item.username }}"
+        host: '127.0.0.1'
+        update_password: "{{ item.update_password }}"
+        password: "{{ test_password1 }}"
+        expect_change: "{{ item.expect_change }}"
+        expect_password_change: "{{ item.expect_change }}"
+        expect_password_hash: "{{ test_password1_hash }}"
+      loop:
+        # all variants set the password when nothing exists
+        - username: test1
+          update_password: always
+          expect_change: true
+        - username: test2
+          update_password: on_create
+          expect_change: true
+        - username: test3
+          update_password: on_new_username
+          expect_change: true
+
+        # assert idempotency
+        - username: test1
+          update_password: always
+          expect_change: false
+        - username: test2
+          update_password: on_create
+          expect_change: false
+        - username: test3
+          update_password: on_new_username
+          expect_change: false
+
+    # same user, new password
+    - include_tasks: assert_user_password.yml
+      vars:
+        username: "{{ item.username }}"
+        host: '127.0.0.1'
+        update_password: "{{ item.update_password }}"
+        password: "{{ test_password2 }}"
+        expect_change: "{{ item.expect_change }}"
+        expect_password_change: "{{ item.expect_change }}"
+        expect_password_hash: "{{ item.expect_password_hash }}"
+      loop:
+        - username: test1
+          update_password: always
+          expect_change: true
+          expect_password_hash: "{{ test_password2_hash }}"
+        - username: test2
+          update_password: on_create
+          expect_change: false
+          expect_password_hash: "{{ test_password1_hash }}"
+        - username: test3
+          update_password: on_new_username
+          expect_change: false
+          expect_password_hash: "{{ test_password1_hash }}"
+
+    # new user, new password
+    - include_tasks: assert_user_password.yml
+      vars:
+        username: "{{ item.username }}"
+        host: '::1'
+        update_password: "{{ item.update_password }}"
+        password: "{{ item.password }}"
+        expect_change: "{{ item.expect_change }}"
+        expect_password_change: "{{ item.expect_password_change }}"
+        expect_password_hash: "{{ item.expect_password_hash }}"
+      loop:
+        - username: test1
+          update_password: always
+          expect_change: true
+          expect_password_change: true
+          password: "{{ test_password1 }}"
+          expect_password_hash: "{{ test_password1_hash }}"
+        - username: test2
+          update_password: on_create
+          expect_change: true
+          expect_password_change: true
+          password: "{{ test_password2 }}"
+          expect_password_hash: "{{ test_password2_hash }}"
+        - username: test3
+          update_password: on_new_username
+          expect_change: true
+          expect_password_change: false
+          password: "{{ test_password2 }}"
+          expect_password_hash: "{{ test_password1_hash }}"
+
+        # prepare for next test: ensure all users have varying passwords
+        - username: test3
+          update_password: always
+          expect_change: true
+          expect_password_change: true
+          password: "{{ test_password2 }}"
+          expect_password_hash: "{{ test_password2_hash }}"
+
+    # another new user, another new password and multiple existing users with varying passwords
+    - include_tasks: assert_user_password.yml
+      vars:
+        username: "{{ item.username }}"
+        host: '2001:db8::1'
+        update_password: "{{ item.update_password }}"
+        password: "{{ test_password3 }}"
+        expect_change: true
+        expect_password_change: true
+        expect_password_hash: "{{ test_password3_hash }}"
+      loop:
+        - username: test1
+          update_password: always
+        - username: test2
+          update_password: on_create
+        - username: test3
+          update_password: on_new_username