Kafka Test Subscriber doesn't match Consumer Group behaviour

[sinnwise]

Currently the Kafka Test Publisher / Subscriber combo returns early upon hitting a subscriber / topic combination.

If an application has two subscribers, across two consumer-groups listening on a single topic, FastStream does not respect that both subscribers should receive a message each from the topic. In testing only the first-matched topic would run.

faststream/faststream/kafka/testing.py

Lines 94 to 102 in ed270ad

	for handler in self.broker._subscribers.values(): # pragma: no branch
	if topic in handler.topics:
	return await call_handler(
	handler=handler,
	message=[incoming]
	if isinstance(handler, AsyncAPIBatchSubscriber)
	else incoming,
	rpc=rpc,
	rpc_timeout=rpc_timeout,

Removing the return and letting each Subscriber handle the alert seems to get me closer to working behaviour (essentially we will have to create a list of subscribers attached to consumer groups and run handle on the first one in each consumer group – not sure how filters should come into play here).

Will work on a PR soon ☺️

[sifex]

PR opened on my non-work account.

[Lancetnik]

Thank you for the Issue and for the PR as well! I'll take a look close soon, sorry for waiting