diff --git a/homeassistant/components/mqtt/debug_info.py b/homeassistant/components/mqtt/debug_info.py
index 13de33923a1..83c78925f56 100644
--- a/homeassistant/components/mqtt/debug_info.py
+++ b/homeassistant/components/mqtt/debug_info.py
@@ -14,7 +14,7 @@ from homeassistant.helpers.typing import DiscoveryInfoType
 from homeassistant.util import dt as dt_util
 
 from .const import ATTR_DISCOVERY_PAYLOAD, ATTR_DISCOVERY_TOPIC
-from .models import DATA_MQTT, MessageCallbackType, PublishPayloadType
+from .models import DATA_MQTT, PublishPayloadType
 
 STORED_MESSAGES = 10
 
@@ -53,41 +53,33 @@ def log_message(
 
 
 def add_subscription(
-    hass: HomeAssistant,
-    message_callback: MessageCallbackType,
-    subscription: str,
-    entity_id: str | None = None,
+    hass: HomeAssistant, subscription: str, entity_id: str | None
 ) -> None:
     """Prepare debug data for subscription."""
-    if not entity_id:
-        entity_id = getattr(message_callback, "__entity_id", None)
     if entity_id:
         entity_info = hass.data[DATA_MQTT].debug_info_entities.setdefault(
             entity_id, {"subscriptions": {}, "discovery_data": {}, "transmitted": {}}
         )
         if subscription not in entity_info["subscriptions"]:
             entity_info["subscriptions"][subscription] = {
-                "count": 0,
+                "count": 1,
                 "messages": deque([], STORED_MESSAGES),
             }
-        entity_info["subscriptions"][subscription]["count"] += 1
+        else:
+            entity_info["subscriptions"][subscription]["count"] += 1
 
 
 def remove_subscription(
-    hass: HomeAssistant,
-    message_callback: MessageCallbackType,
-    subscription: str,
-    entity_id: str | None = None,
+    hass: HomeAssistant, subscription: str, entity_id: str | None
 ) -> None:
     """Remove debug data for subscription if it exists."""
-    if not entity_id:
-        entity_id = getattr(message_callback, "__entity_id", None)
     if entity_id and entity_id in (
         debug_info_entities := hass.data[DATA_MQTT].debug_info_entities
     ):
-        debug_info_entities[entity_id]["subscriptions"][subscription]["count"] -= 1
-        if not debug_info_entities[entity_id]["subscriptions"][subscription]["count"]:
-            debug_info_entities[entity_id]["subscriptions"].pop(subscription)
+        subscriptions = debug_info_entities[entity_id]["subscriptions"]
+        subscriptions[subscription]["count"] -= 1
+        if not subscriptions[subscription]["count"]:
+            subscriptions.pop(subscription)
 
 
 def add_entity_discovery_data(
diff --git a/homeassistant/components/mqtt/subscription.py b/homeassistant/components/mqtt/subscription.py
index 40f9f130134..3f3f67970f3 100644
--- a/homeassistant/components/mqtt/subscription.py
+++ b/homeassistant/components/mqtt/subscription.py
@@ -15,7 +15,7 @@ from .const import DEFAULT_QOS
 from .models import MessageCallbackType
 
 
-@dataclass(slots=True)
+@dataclass(slots=True, kw_only=True)
 class EntitySubscription:
     """Class to hold data about an active entity topic subscription."""
 
@@ -26,8 +26,8 @@ class EntitySubscription:
     unsubscribe_callback: Callable[[], None] | None
     qos: int = 0
     encoding: str = "utf-8"
-    entity_id: str | None = None
-    job_type: HassJobType | None = None
+    entity_id: str | None
+    job_type: HassJobType | None
 
     def resubscribe_if_necessary(
         self, hass: HomeAssistant, other: EntitySubscription | None
@@ -42,18 +42,14 @@ class EntitySubscription:
         if other is not None and other.unsubscribe_callback is not None:
             other.unsubscribe_callback()
             # Clear debug data if it exists
-            debug_info.remove_subscription(
-                self.hass, other.message_callback, str(other.topic), other.entity_id
-            )
+            debug_info.remove_subscription(self.hass, str(other.topic), other.entity_id)
 
         if self.topic is None:
             # We were asked to remove the subscription or not to create it
             return
 
         # Prepare debug data
-        debug_info.add_subscription(
-            self.hass, self.message_callback, self.topic, self.entity_id
-        )
+        debug_info.add_subscription(self.hass, self.topic, self.entity_id)
 
         self.should_subscribe = True
 
@@ -110,15 +106,15 @@ def async_prepare_subscribe_topics(
     for key, value in topics.items():
         # Extract the new requested subscription
         requested = EntitySubscription(
-            topic=value.get("topic", None),
-            message_callback=value.get("msg_callback", None),
+            topic=value.get("topic"),
+            message_callback=value["msg_callback"],
             unsubscribe_callback=None,
             qos=value.get("qos", DEFAULT_QOS),
             encoding=value.get("encoding", "utf-8"),
             hass=hass,
             should_subscribe=None,
-            entity_id=value.get("entity_id", None),
-            job_type=value.get("job_type", None),
+            entity_id=value.get("entity_id"),
+            job_type=value.get("job_type"),
         )
         # Get the current subscription state
         current = current_subscriptions.pop(key, None)
@@ -132,7 +128,6 @@ def async_prepare_subscribe_topics(
             # Clear debug data if it exists
             debug_info.remove_subscription(
                 hass,
-                remaining.message_callback,
                 str(remaining.topic),
                 remaining.entity_id,
             )