diff --git a/homeassistant/helpers/integration_platform.py b/homeassistant/helpers/integration_platform.py
index 9255824cddf..ef05dae518b 100644
--- a/homeassistant/helpers/integration_platform.py
+++ b/homeassistant/helpers/integration_platform.py
@@ -8,8 +8,8 @@ import logging
 from typing import Any
 
 from homeassistant.const import EVENT_COMPONENT_LOADED
-from homeassistant.core import Event, HomeAssistant
-from homeassistant.loader import async_get_integration, bind_hass
+from homeassistant.core import Event, HomeAssistant, callback
+from homeassistant.loader import Integration, async_get_integrations, bind_hass
 from homeassistant.setup import ATTR_COMPONENT
 
 _LOGGER = logging.getLogger(__name__)
@@ -26,14 +26,24 @@ class IntegrationPlatform:
 
 
 async def _async_process_single_integration_platform_component(
-    hass: HomeAssistant, component_name: str, integration_platform: IntegrationPlatform
+    hass: HomeAssistant,
+    component_name: str,
+    integration: Integration | Exception,
+    integration_platform: IntegrationPlatform,
 ) -> None:
     """Process a single integration platform."""
     if component_name in integration_platform.seen_components:
         return
     integration_platform.seen_components.add(component_name)
 
-    integration = await async_get_integration(hass, component_name)
+    if isinstance(integration, Exception):
+        _LOGGER.exception(
+            "Error importing integration %s for %s",
+            component_name,
+            integration_platform.platform_name,
+        )
+        return
+
     platform_name = integration_platform.platform_name
 
     try:
@@ -75,14 +85,22 @@ async def async_process_integration_platform_for_component(
     integration_platforms: list[IntegrationPlatform] = hass.data[
         DATA_INTEGRATION_PLATFORMS
     ]
-    await asyncio.gather(
-        *[
+    integrations = await async_get_integrations(hass, (component_name,))
+    tasks = [
+        asyncio.create_task(
             _async_process_single_integration_platform_component(
-                hass, component_name, integration_platform
-            )
-            for integration_platform in integration_platforms
-        ]
-    )
+                hass,
+                component_name,
+                integrations[component_name],
+                integration_platform,
+            ),
+            name=f"process integration platform {integration_platform.platform_name} for {component_name}",
+        )
+        for integration_platform in integration_platforms
+        if component_name not in integration_platform.seen_components
+    ]
+    if tasks:
+        await asyncio.gather(*tasks)
 
 
 @bind_hass
@@ -98,25 +116,39 @@ async def async_process_integration_platforms(
 
         async def _async_component_loaded(event: Event) -> None:
             """Handle a new component loaded."""
-            comp = event.data[ATTR_COMPONENT]
-            if "." not in comp:
-                await async_process_integration_platform_for_component(hass, comp)
+            await async_process_integration_platform_for_component(
+                hass, event.data[ATTR_COMPONENT]
+            )
 
-        hass.bus.async_listen(EVENT_COMPONENT_LOADED, _async_component_loaded)
+        @callback
+        def _async_component_loaded_filter(event: Event) -> bool:
+            """Handle integration platforms loaded."""
+            return "." not in event.data[ATTR_COMPONENT]
+
+        hass.bus.async_listen(
+            EVENT_COMPONENT_LOADED,
+            _async_component_loaded,
+            event_filter=_async_component_loaded_filter,
+        )
 
     integration_platforms: list[IntegrationPlatform] = hass.data[
         DATA_INTEGRATION_PLATFORMS
     ]
     integration_platform = IntegrationPlatform(platform_name, process_platform, set())
     integration_platforms.append(integration_platform)
-    if top_level_components := (
+    if top_level_components := [
         comp for comp in hass.config.components if "." not in comp
-    ):
-        await asyncio.gather(
-            *[
+    ]:
+        integrations = await async_get_integrations(hass, top_level_components)
+        tasks = [
+            asyncio.create_task(
                 _async_process_single_integration_platform_component(
-                    hass, comp, integration_platform
-                )
-                for comp in top_level_components
-            ]
-        )
+                    hass, comp, integrations[comp], integration_platform
+                ),
+                name=f"process integration platform {platform_name} for {comp}",
+            )
+            for comp in top_level_components
+            if comp not in integration_platform.seen_components
+        ]
+        if tasks:
+            await asyncio.gather(*tasks)
diff --git a/tests/helpers/test_integration_platform.py b/tests/helpers/test_integration_platform.py
index 5848c6f9541..2dfc0742e26 100644
--- a/tests/helpers/test_integration_platform.py
+++ b/tests/helpers/test_integration_platform.py
@@ -1,6 +1,8 @@
 """Test integration platform helpers."""
 from unittest.mock import Mock
 
+import pytest
+
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.integration_platform import (
     async_process_integration_platform_for_component,
@@ -51,3 +53,27 @@ async def test_process_integration_platforms_none_loaded(hass: HomeAssistant) ->
     # Verify we can call async_process_integration_platform_for_component
     # when there are none loaded and it does not throw
     await async_process_integration_platform_for_component(hass, "any")
+
+
+async def test_broken_integration(
+    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Test handling an integration with a broken or missing manifest."""
+    Mock()
+    hass.config.components.add("loaded")
+
+    event_platform = Mock()
+    mock_platform(hass, "event.platform_to_check", event_platform)
+
+    processed = []
+
+    async def _process_platform(hass, domain, platform):
+        """Process platform."""
+        processed.append((domain, platform))
+
+    await async_process_integration_platforms(
+        hass, "platform_to_check", _process_platform
+    )
+
+    assert len(processed) == 0
+    assert "Error importing integration loaded for platform_to_check" in caplog.text