diff --git a/homeassistant/components/mqtt/__init__.py b/homeassistant/components/mqtt/__init__.py
index 306132c4f36..62aad6ca7fe 100644
--- a/homeassistant/components/mqtt/__init__.py
+++ b/homeassistant/components/mqtt/__init__.py
@@ -248,20 +248,7 @@ async def _async_config_entry_updated(hass: HomeAssistant, entry: ConfigEntry) -
 
     Causes for this is config entry options changing.
     """
-    mqtt_data = get_mqtt_data(hass)
-    assert (client := mqtt_data.client) is not None
-
-    if (conf := mqtt_data.config) is None:
-        conf = CONFIG_SCHEMA_BASE(dict(entry.data))
-
-    mqtt_data.config = _merge_extended_config(entry, conf)
-    await client.async_disconnect()
-    client.init_client()
-    await client.async_connect()
-
-    await discovery.async_stop(hass)
-    if client.conf.get(CONF_DISCOVERY):
-        await _async_setup_discovery(hass, cast(ConfigType, mqtt_data.config), entry)
+    await hass.config_entries.async_reload(entry.entry_id)
 
 
 async def async_fetch_config(hass: HomeAssistant, entry: ConfigEntry) -> dict | None:
@@ -317,7 +304,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     if mqtt_data.subscriptions_to_restore:
         mqtt_data.client.subscriptions = mqtt_data.subscriptions_to_restore
         mqtt_data.subscriptions_to_restore = []
-    entry.add_update_listener(_async_config_entry_updated)
+    mqtt_data.reload_dispatchers.append(
+        entry.add_update_listener(_async_config_entry_updated)
+    )
 
     await mqtt_data.client.async_connect()
 
diff --git a/homeassistant/components/mqtt/config_flow.py b/homeassistant/components/mqtt/config_flow.py
index afa2d98af2b..55fd68f8b6a 100644
--- a/homeassistant/components/mqtt/config_flow.py
+++ b/homeassistant/components/mqtt/config_flow.py
@@ -262,7 +262,9 @@ class MQTTOptionsFlowHandler(config_entries.OptionsFlow):
                 updated_config.update(self.broker_config)
                 updated_config.update(options_config)
                 self.hass.config_entries.async_update_entry(
-                    self.config_entry, data=updated_config
+                    self.config_entry,
+                    data=updated_config,
+                    title=str(self.broker_config[CONF_BROKER]),
                 )
                 return self.async_create_entry(title="", data={})
 
diff --git a/tests/components/mqtt/test_config_flow.py b/tests/components/mqtt/test_config_flow.py
index dba06e5cd5b..2ad216a3b7b 100644
--- a/tests/components/mqtt/test_config_flow.py
+++ b/tests/components/mqtt/test_config_flow.py
@@ -1,5 +1,5 @@
 """Test config flow."""
-from unittest.mock import patch
+from unittest.mock import AsyncMock, patch
 
 import pytest
 import voluptuous as vol
@@ -23,6 +23,15 @@ def mock_finish_setup():
         yield mock_finish
 
 
+@pytest.fixture
+def mock_reload_after_entry_update():
+    """Mock out the reload after updating the entry."""
+    with patch(
+        "homeassistant.components.mqtt._async_config_entry_updated"
+    ) as mock_reload:
+        yield mock_reload
+
+
 @pytest.fixture
 def mock_try_connection():
     """Mock the try connection method."""
@@ -180,6 +189,8 @@ async def test_manual_config_set(
     mock_try_connection.assert_called_once_with(hass, "127.0.0.1", 1883, None, None)
     # Check config entry got setup
     assert len(mock_finish_setup.mock_calls) == 1
+    config_entry = hass.config_entries.async_entries(mqtt.DOMAIN)[0]
+    assert config_entry.title == "127.0.0.1"
 
 
 async def test_user_single_instance(hass):
@@ -269,7 +280,16 @@ async def test_hassio_confirm(hass, mock_try_connection_success, mock_finish_set
     assert len(mock_finish_setup.mock_calls) == 1
 
 
-async def test_option_flow(hass, mqtt_mock_entry_no_yaml_config, mock_try_connection):
+@patch(
+    "homeassistant.config.async_hass_config_yaml",
+    AsyncMock(return_value={}),
+)
+async def test_option_flow(
+    hass,
+    mqtt_mock_entry_no_yaml_config,
+    mock_try_connection,
+    mock_reload_after_entry_update,
+):
     """Test config flow options."""
     mqtt_mock = await mqtt_mock_entry_no_yaml_config()
     mock_try_connection.return_value = True
@@ -339,11 +359,16 @@ async def test_option_flow(hass, mqtt_mock_entry_no_yaml_config, mock_try_connec
     }
 
     await hass.async_block_till_done()
-    assert mqtt_mock.async_connect.call_count == 1
+    assert config_entry.title == "another-broker"
+    # assert that the entry was reloaded with the new config
+    assert mock_reload_after_entry_update.call_count == 1
 
 
 async def test_disable_birth_will(
-    hass, mqtt_mock_entry_no_yaml_config, mock_try_connection
+    hass,
+    mqtt_mock_entry_no_yaml_config,
+    mock_try_connection,
+    mock_reload_after_entry_update,
 ):
     """Test disabling birth and will."""
     mqtt_mock = await mqtt_mock_entry_no_yaml_config()
@@ -404,7 +429,8 @@ async def test_disable_birth_will(
     }
 
     await hass.async_block_till_done()
-    assert mqtt_mock.async_connect.call_count == 1
+    # assert that the entry was reloaded with the new config
+    assert mock_reload_after_entry_update.call_count == 1
 
 
 def get_default(schema, key):
@@ -426,7 +452,10 @@ def get_suggested(schema, key):
 
 
 async def test_option_flow_default_suggested_values(
-    hass, mqtt_mock_entry_no_yaml_config, mock_try_connection_success
+    hass,
+    mqtt_mock_entry_no_yaml_config,
+    mock_try_connection_success,
+    mock_reload_after_entry_update,
 ):
     """Test config flow options has default/suggested values."""
     await mqtt_mock_entry_no_yaml_config()