Feature: Add ability to specify on_assign, on_revoke, and on_lost callbacks for a Confluent subscriber

[andreaimprovised]

Is your feature request related to a problem? Please describe.
Yes, I want to know when my confluent Consumer gets topic partitions assigned and removed.

Currently, I reach through FastStream into confluent_kafka.Consumer.assignment() every time my k8s liveness probe runs, but its noisy and most notably, not right when it happens.

I may even, at times, want to do something with the informaiton beyond logging. Potentially clear some cached state, cancel some running threads/processes, etc...

Describe the solution you'd like

I want to specify at the subscriber registration level the callbacks that I want called, and for FastStream to pass them into the

confluent_kafka.Consumer.subscribe() call inside AsyncConfluentConsumer.

Feature code example

from faststream import FastStream

...

broker = KafkaBroker(...)

@broker.subscriber(
    "my-topic",
    on_assign=lambda consumer, partitions: ...,
    on_revoke=lambda consumer, partitions: ...,
)
def my_handler(body: str):
    print(body)

Describe alternatives you've considered

I monkey patch AsyncConfluentConsumer at import time in the FastStream library.

import faststream.confluent.broker.broker
from faststream.confluent.broker.broker import AsyncConfluentConsumer
from observing.observing import logger


class PatchedAsyncConfluentConsumer(AsyncConfluentConsumer):
    """A patched version of the AsyncConfluentConsumer class."""

    def __init__(self, *topics, **kwargs):
        super().__init__(*topics, **kwargs)
        self.topic_partitions = set()

    def on_revoke(self, consumer, partitions):
        """Conditionally pauses the consumer when partitions are revoked."""
        self.topic_partitions -= set(partitions)
        logger.info(
            "Consumer rebalance event: partitions revoked.",
            topic_partitions=dict(
                n_revoked=len(partitions),
                revoked=[
                    dict(topic=tp.topic, partition=tp.partition) for tp in partitions
                ],
                n_current=len(self.topic_partitions),
                current=[
                    dict(topic=tp.topic, partition=tp.partition)
                    for tp in self.topic_partitions
                ],
            ),
            memberid=self.consumer.memberid(),
            topics=self.topics,
            config=dict(
                group_id=self.config.get("group.id"),
                group_instance_id=self.config.get("group.instance.id"),
            ),
        )

    def on_assign(self, consumer, partitions):
        """Conditionally resumes the consumer when partitions are assigned."""
        self.topic_partitions |= set(partitions)
        logger.info(
            "Consumer rebalance event: partitions assigned.",
            topic_partitions=dict(
                n_assigned=len(partitions),
                assigned=[
                    dict(topic=tp.topic, partition=tp.partition) for tp in partitions
                ],
                n_current=len(self.topic_partitions),
                current=[
                    dict(topic=tp.topic, partition=tp.partition)
                    for tp in self.topic_partitions
                ],
            ),
            memberid=self.consumer.memberid(),
            topics=self.topics,
            config=dict(
                group_id=self.config.get("group.id"),
                group_instance_id=self.config.get("group.instance.id"),
            ),
        )

    async def start(self) -> None:
        """Starts the Kafka consumer and subscribes to the specified topics."""
        self.consumer.subscribe(
            self.topics, on_revoke=self.on_revoke, on_assign=self.on_assign
        )


def patch_async_confluent_consumer():
    logger.info("Patching AsyncConfluentConsumer.")
    faststream.confluent.broker.broker.AsyncConfluentConsumer = (
        PatchedAsyncConfluentConsumer
    )

Obviously, this is ideal for no one.

Additional context