diff --git a/homeassistant/components/mqtt/client.py b/homeassistant/components/mqtt/client.py
index 22833183b69..a16f7f4b9c5 100644
--- a/homeassistant/components/mqtt/client.py
+++ b/homeassistant/components/mqtt/client.py
@@ -495,6 +495,9 @@ class MQTT:
         mqttc.on_subscribe = self._async_mqtt_on_callback
         mqttc.on_unsubscribe = self._async_mqtt_on_callback
 
+        # suppress exceptions at callback
+        mqttc.suppress_exceptions = True
+
         if will := self.conf.get(CONF_WILL_MESSAGE, DEFAULT_WILL):
             will_message = PublishMessage(**will)
             mqttc.will_set(
@@ -989,10 +992,21 @@ class MQTT:
     def _async_mqtt_on_message(
         self, _mqttc: mqtt.Client, _userdata: None, msg: mqtt.MQTTMessage
     ) -> None:
-        topic = msg.topic
-        # msg.topic is a property that decodes the topic to a string
-        # every time it is accessed. Save the result to avoid
-        # decoding the same topic multiple times.
+        try:
+            # msg.topic is a property that decodes the topic to a string
+            # every time it is accessed. Save the result to avoid
+            # decoding the same topic multiple times.
+            topic = msg.topic
+        except UnicodeDecodeError:
+            bare_topic: bytes = getattr(msg, "_topic")
+            _LOGGER.warning(
+                "Skipping received%s message on invalid topic %s (qos=%s): %s",
+                " retained" if msg.retain else "",
+                bare_topic,
+                msg.qos,
+                msg.payload[0:8192],
+            )
+            return
         _LOGGER.debug(
             "Received%s message on %s (qos=%s): %s",
             " retained" if msg.retain else "",
diff --git a/tests/components/mqtt/test_init.py b/tests/components/mqtt/test_init.py
index a9f4a9f7454..938426d48ed 100644
--- a/tests/components/mqtt/test_init.py
+++ b/tests/components/mqtt/test_init.py
@@ -8,8 +8,9 @@ import json
 import logging
 import socket
 import ssl
+import time
 from typing import Any, TypedDict
-from unittest.mock import ANY, MagicMock, call, mock_open, patch
+from unittest.mock import ANY, MagicMock, Mock, call, mock_open, patch
 
 from freezegun.api import FrozenDateTimeFactory
 import paho.mqtt.client as paho_mqtt
@@ -951,6 +952,42 @@ async def test_receiving_non_utf8_message_gets_logged(
     )
 
 
+async def test_receiving_message_with_non_utf8_topic_gets_logged(
+    hass: HomeAssistant,
+    mqtt_mock_entry: MqttMockHAClientGenerator,
+    record_calls: MessageCallbackType,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Test receiving a non utf8 encoded topic."""
+    await mqtt_mock_entry()
+    await mqtt.async_subscribe(hass, "test-topic", record_calls)
+
+    # Local import to avoid processing MQTT modules when running a testcase
+    # which does not use MQTT.
+
+    # pylint: disable-next=import-outside-toplevel
+    from paho.mqtt.client import MQTTMessage
+
+    # pylint: disable-next=import-outside-toplevel
+    from homeassistant.components.mqtt.models import MqttData
+
+    msg = MQTTMessage(topic=b"tasmota/discovery/18FE34E0B760\xcc\x02")
+    msg.payload = b"Payload"
+    msg.qos = 2
+    msg.retain = True
+    msg.timestamp = time.monotonic()
+
+    mqtt_data: MqttData = hass.data["mqtt"]
+    assert mqtt_data.client
+    mqtt_data.client._async_mqtt_on_message(Mock(), None, msg)
+
+    assert (
+        "Skipping received retained message on invalid "
+        "topic b'tasmota/discovery/18FE34E0B760\\xcc\\x02' "
+        "(qos=2): b'Payload'" in caplog.text
+    )
+
+
 async def test_all_subscriptions_run_when_decode_fails(
     hass: HomeAssistant,
     mqtt_mock_entry: MqttMockHAClientGenerator,