feature: manual partition assignment to Kafka

[spataphore1337]

Description

This feature allows you to assign specific partitions in a topic to which your Kafka concumer needs to subscribe.

from faststream.kafka import TopicPartition
from faststream import FastStream
from faststream.kafka import KafkaRouter

router = KafkaRouter()

topic_partition_fisrt = TopicPartition("my_topic", 1)
topic_partition_second = TopicPartition("my_topic", 2)

@router.subscribe(partition=[topic_partition_fisrt, topic_partition_second])
async def some_consumer(msg):
   ...

Type of change

Please delete options that are not relevant.

• Documentation (typos, code examples, or any documentation updates)
• Bug fix (a non-breaking change that resolves an issue)
• New feature (a non-breaking change that adds functionality)
• Breaking change (a fix or feature that would disrupt existing functionality)
• This change requires a documentation update

Checklist

• My code adheres to the style guidelines of this project (scripts/lint.sh shows no errors)
• I have conducted a self-review of my own code
• I have made the necessary changes to the documentation
• My changes do not generate any new warnings
• I have added tests to validate the effectiveness of my fix or the functionality of my new feature
• Both new and existing unit tests pass successfully on my local environment by running scripts/test-cov.sh
• I have ensured that static analysis tests are passing by running scripts/static-anaylysis.sh
• I have included code examples to illustrate the modifications

[Lancetnik]

@andrijadukic can you help us a bit? For some reason the following example doesn't work (handler doesn't see any message)

from faststream import FastStream, Logger
from faststream.kafka import KafkaBroker, TopicPartition

broker = KafkaBroker()
app = FastStream(broker)

@broker.subscriber(partitions=[TopicPartition("test", 0)])
async def handler(msg, logger: Logger):
    logger.info(msg)

@app.after_startup
async def t():
    await broker.publish("Hi", "test")
    await broker.publish("again", "test", partition=0)

We are using just aiokafka consumer.assign method. Should we use something extra to make it working?

https://aiokafka.readthedocs.io/en/stable/consumer.html?highlight=subscribe#manual-partition-assignment