diff --git a/changelogs/fragments/66252-mysql_replication_fail_on_error.yml b/changelogs/fragments/66252-mysql_replication_fail_on_error.yml
new file mode 100644
index 00000000000..cd72f4cbf2e
--- /dev/null
+++ b/changelogs/fragments/66252-mysql_replication_fail_on_error.yml
@@ -0,0 +1,2 @@
+minor_changes:
+- mysql_replication - add ``fail_on_error`` parameter (https://github.com/ansible/ansible/pull/66252).
diff --git a/lib/ansible/modules/database/mysql/mysql_replication.py b/lib/ansible/modules/database/mysql/mysql_replication.py
index a4019d714cb..93620b60d7b 100644
--- a/lib/ansible/modules/database/mysql/mysql_replication.py
+++ b/lib/ansible/modules/database/mysql/mysql_replication.py
@@ -146,6 +146,12 @@ options:
     - For more information see U(https://dev.mysql.com/doc/refman/8.0/en/replication-multi-source.html).
     type: str
     version_added: '2.10'
+  fail_on_error:
+    description:
+    - Fails on error when calling mysql.
+    type: bool
+    default: False
+    version_added: '2.10'
 
 notes:
 - If an empty value for the parameter of string type is needed, use an empty string.
@@ -211,6 +217,18 @@ EXAMPLES = r'''
     and reset the binary log index file on the master
   mysql_replication:
     mode: resetmaster
+
+- name: Run start slave and fail the task on errors
+  mysql_replication:
+    mode: startslave
+    connection_name: master-1
+    fail_on_error: yes
+
+- name: Change master and fail on error (like when slave thread is running)
+  mysql_replication:
+    mode: changemaster
+    fail_on_error: yes
+
 '''
 
 RETURN = r'''
@@ -252,7 +270,7 @@ def get_slave_status(cursor, connection_name='', channel=''):
     return slavestatus
 
 
-def stop_slave(cursor, connection_name='', channel=''):
+def stop_slave(module, cursor, connection_name='', channel='', fail_on_error=False):
     if connection_name:
         query = "STOP SLAVE '%s'" % connection_name
     else:
@@ -265,12 +283,16 @@ def stop_slave(cursor, connection_name='', channel=''):
         executed_queries.append(query)
         cursor.execute(query)
         stopped = True
-    except Exception:
+    except mysql_driver.Warning as e:
+        stopped = False
+    except Exception as e:
+        if fail_on_error:
+            module.fail_json(msg="STOP SLAVE failed: %s" % to_native(e))
         stopped = False
     return stopped
 
 
-def reset_slave(cursor, connection_name='', channel=''):
+def reset_slave(module, cursor, connection_name='', channel='', fail_on_error=False):
     if connection_name:
         query = "RESET SLAVE '%s'" % connection_name
     else:
@@ -283,12 +305,16 @@ def reset_slave(cursor, connection_name='', channel=''):
         executed_queries.append(query)
         cursor.execute(query)
         reset = True
-    except Exception:
+    except mysql_driver.Warning as e:
+        reset = False
+    except Exception as e:
+        if fail_on_error:
+            module.fail_json(msg="RESET SLAVE failed: %s" % to_native(e))
         reset = False
     return reset
 
 
-def reset_slave_all(cursor, connection_name='', channel=''):
+def reset_slave_all(module, cursor, connection_name='', channel='', fail_on_error=False):
     if connection_name:
         query = "RESET SLAVE '%s' ALL" % connection_name
     else:
@@ -301,23 +327,31 @@ def reset_slave_all(cursor, connection_name='', channel=''):
         executed_queries.append(query)
         cursor.execute(query)
         reset = True
-    except Exception:
+    except mysql_driver.Warning as e:
+        reset = False
+    except Exception as e:
+        if fail_on_error:
+            module.fail_json(msg="RESET SLAVE ALL failed: %s" % to_native(e))
         reset = False
     return reset
 
 
-def reset_master(cursor):
+def reset_master(module, cursor, fail_on_error=False):
     query = 'RESET MASTER'
     try:
         executed_queries.append(query)
         cursor.execute(query)
         reset = True
-    except Exception:
+    except mysql_driver.Warning as e:
+        reset = False
+    except Exception as e:
+        if fail_on_error:
+            module.fail_json(msg="RESET MASTER failed: %s" % to_native(e))
         reset = False
     return reset
 
 
-def start_slave(cursor, connection_name='', channel=''):
+def start_slave(module, cursor, connection_name='', channel='', fail_on_error=False):
     if connection_name:
         query = "START SLAVE '%s'" % connection_name
     else:
@@ -330,7 +364,11 @@ def start_slave(cursor, connection_name='', channel=''):
         executed_queries.append(query)
         cursor.execute(query)
         started = True
-    except Exception:
+    except mysql_driver.Warning as e:
+        started = False
+    except Exception as e:
+        if fail_on_error:
+            module.fail_json(msg="START SLAVE failed: %s" % to_native(e))
         started = False
     return started
 
@@ -384,6 +422,7 @@ def main():
             master_delay=dict(type='int'),
             connection_name=dict(type='str'),
             channel=dict(type='str'),
+            fail_on_error=dict(type='bool', default=False),
         ),
         mutually_exclusive=[
             ['connection_name', 'channel']
@@ -418,6 +457,7 @@ def main():
         master_use_gtid = module.params["master_use_gtid"]
     connection_name = module.params["connection_name"]
     channel = module.params['channel']
+    fail_on_error = module.params['fail_on_error']
 
     if mysql_driver is None:
         module.fail_json(msg=mysql_driver_fail_msg)
@@ -502,31 +542,31 @@ def main():
         result['changed'] = True
         module.exit_json(queries=executed_queries, **result)
     elif mode in "startslave":
-        started = start_slave(cursor, connection_name, channel)
+        started = start_slave(module, cursor, connection_name, channel, fail_on_error)
         if started is True:
             module.exit_json(msg="Slave started ", changed=True, queries=executed_queries)
         else:
             module.exit_json(msg="Slave already started (Or cannot be started)", changed=False, queries=executed_queries)
     elif mode in "stopslave":
-        stopped = stop_slave(cursor, connection_name, channel)
+        stopped = stop_slave(module, cursor, connection_name, channel, fail_on_error)
         if stopped is True:
             module.exit_json(msg="Slave stopped", changed=True, queries=executed_queries)
         else:
             module.exit_json(msg="Slave already stopped", changed=False, queries=executed_queries)
     elif mode in "resetmaster":
-        reset = reset_master(cursor)
+        reset = reset_master(module, cursor, fail_on_error)
         if reset is True:
             module.exit_json(msg="Master reset", changed=True, queries=executed_queries)
         else:
             module.exit_json(msg="Master already reset", changed=False, queries=executed_queries)
     elif mode in "resetslave":
-        reset = reset_slave(cursor, connection_name, channel)
+        reset = reset_slave(module, cursor, connection_name, channel, fail_on_error)
         if reset is True:
             module.exit_json(msg="Slave reset", changed=True, queries=executed_queries)
         else:
             module.exit_json(msg="Slave already reset", changed=False, queries=executed_queries)
     elif mode in "resetslaveall":
-        reset = reset_slave_all(cursor, connection_name, channel)
+        reset = reset_slave_all(module, cursor, connection_name, channel, fail_on_error)
         if reset is True:
             module.exit_json(msg="Slave reset", changed=True, queries=executed_queries)
         else:
diff --git a/test/integration/targets/mysql_replication/tasks/mysql_replication_initial.yml b/test/integration/targets/mysql_replication/tasks/mysql_replication_initial.yml
index ce7c260c1e4..f0c8f1e7bb2 100644
--- a/test/integration/targets/mysql_replication/tasks/mysql_replication_initial.yml
+++ b/test/integration/targets/mysql_replication/tasks/mysql_replication_initial.yml
@@ -13,7 +13,7 @@
     name: '{{ test_db }}'
 
 - name: Dump all databases from the master
-  shell: 'mysqldump -P {{ master_port }} -h 127.0.01 --all-databases --master-data=2 > {{ dump_path }}'
+  shell: 'mysqldump -P {{ master_port }} -h 127.0.0.1 --all-databases --master-data=2 > {{ dump_path }}'
 
 - name: Restore the dump to the standby
   shell: 'mysql -P {{ standby_port }} -h 127.0.0.1 < {{ dump_path }}'
@@ -32,6 +32,46 @@
     - master_status.Position != 0
     - master_status is not changed
 
+# Test startslave fails without changemaster first. This needs fail_on_error
+- name: Start slave and fail because master is not specified; failing on error as requested
+  mysql_replication:
+    login_host: 127.0.0.1
+    login_port: "{{ standby_port }}"
+    mode: startslave
+    fail_on_error: yes
+  register: result
+  ignore_errors: yes
+
+- assert:
+    that:
+    - result is failed
+
+# Test startslave doesn't fail if fail_on_error: no
+- name: Start slave and fail without propagating it to ansible as we were asked not to
+  mysql_replication:
+    login_host: 127.0.0.1
+    login_port: "{{ standby_port }}"
+    mode: startslave
+    fail_on_error: no
+  register: result
+
+- assert:
+    that:
+    - result is not failed
+
+# Test startslave doesn't fail if there is no fail_on_error.
+# This is suboptimal because nothing happens, but it's the old behavior.
+- name: Start slave and fail without propagating it to ansible as previous versions did not fail on error
+  mysql_replication:
+    login_host: 127.0.0.1
+    login_port: "{{ standby_port }}"
+    mode: startslave
+  register: result
+
+- assert:
+    that:
+    - result is not failed
+
 # Test changemaster mode:
 # master_ssl_ca will be set as '' to check the module's behaviour for #23976,
 # must be converted to an empty string
@@ -112,6 +152,18 @@
     that:
     - slave_status.Exec_Master_Log_Pos != master_status.Position
 
+- name: Start slave that is already running
+  mysql_replication:
+    login_host: 127.0.0.1
+    login_port: "{{ standby_port }}"
+    mode: startslave
+    fail_on_error: true
+  register: result
+
+- assert:
+    that:
+    - result is not changed
+
 # Test stopslave mode:
 - name: Stop slave
   mysql_replication:
@@ -124,3 +176,16 @@
     that:
     - result is changed
     - result.queries == ["STOP SLAVE"]
+
+# Test stopslave mode:
+- name: Stop slave that is no longer running
+  mysql_replication:
+    login_host: 127.0.0.1
+    login_port: "{{ standby_port }}"
+    mode: stopslave
+    fail_on_error: true
+  register: result
+
+- assert:
+    that:
+    - result is not changed