Merged in from Main

faststream/kafka/testing.py

@@ -108,13 +108,17 @@ async def publish(  # type: ignore[override]
         return_value = None
 
         for handler in self.broker._subscribers.values():  # pragma: no branch
+            call = False
 
-            # for p in handler.partitions:
-            #     if p.topic == topic and (partition is None or p.partition == partition):
-            #         call = True
+            for p in handler.partitions:
+                if p.topic == topic and (partition is None or p.partition == partition):
+                    call = True
 
             if topic in handler.topics:
-                return await call_handler(
+                call = True
+
+            if call:
+                handle_value = await call_handler(
                     handler=handler,
                     message=[incoming]
                     if isinstance(handler, AsyncAPIBatchSubscriber)