Filter Kafka Batch

[dom-gunstone]

Hi,

Is it possible to filter a (Confluent) Kafka batch like this?

from faststream.confluent import KafkaBroker

broker = KafkaBroker(...)
subscriber = broker.subscriber(..., batch=True)

@broker.subscriber(filter=lambda msg: msg.headers["user"] == "foo")
async def foo(body: list[dict]) -> None:
    # Only messages with headers["user"] == "foo"
    ...
    

@broker.subscriber(filter=lambda msg: msg.headers["user"] == "bar")
async def foo(body: list[dict]) -> None: 
    # Only messages with headers["user"] == "bar"
    ...

[Lancetnik]

@dom-gunstone unfortunately, you can't filter out batch between various handlers. But it looks like a good feature

[dom-gunstone]

Thank you!