Possibility of using manual partition assignment instead of consumer groups in KafkaBroker

[andrijadukic]

Hi!
One feature that I don't think is available right now, but might be useful in certain use cases, is to provide a way to do manual partition assignment instead of simply using consumer groups.
Using confluent-kafka, it would look something like this:

consumer = Consumer(config)
topics = consumer.list_topics().topics
partitions = [TopicPartition(target_topic, partition=partition, offset=OFFSET_END) for partition in list(topics[target_topic].partitions.keys())]
consumer.assign(partitions)

In the case this isn't supported right now (by settings the arguments a certain way), faststream could potentially simply things here to something like this (assign doesn't exist, but could be a friendlier way to add this functionality than adding new arguments to subscribe):

@broker.assign(topics="some-topic")
def handle1(message: str):
    do_something()

@broker.assign(partitions=[TopicPartition(topic="some-topic", partition=0, offset=OFFSET_END), TopicPartition(topic="some-topic", partition=1, offset=OFFSET_END)])
def handle2(message: str):
    do_something()

An example of where this would be useful is in any case where each consumer application instance needs to process the entire topic, so pretty much the opposite of what consumer groups are used for.

A workaround that comes to mind is to set group_id to some random value that's unique to each running instance, but this is a bit hacky, and could potentially still count as a consumer group from the perspective of the broker, which is an issue in certain Kafka offerings where the total number of consumer groups is limited to some small amount, so it's a shame to waste it on this.

Here's a link to this exact discussion in the confluent-kafka repo.

Great library by the way 👏

[Lancetnik]

Hi! We are planning to add this option in #1319 for confluent and aio-kafka brokers both

It should looks like

@broker.subscriber(partitions=[
    TopicPartition(topic="some-topic", partition=0, offset=OFFSET_END),
    TopicPartition(topic="some-topic", partition=1, offset=OFFSET_END),
])
def handle2(message: str):
    do_something()

We are using @broker.subscriber specific options for all ways to subscribe on events

[andrijadukic]

Awesome, sounds great!