feat: attach the rebalance listener to concurrent subscribers automatically

[lesnik512]

Problem

ConsumerRebalanceListener is the only thing that commits finished work when partitions are revoked, but nothing attached it:

• FastStream passes a rebalance listener to aiokafka only when the subscriber was declared with listener=; otherwise it installs just its own logging listener (faststream/kafka/subscriber/usecase.py).
• handler.create_rebalance_listener() needs the KafkaConcurrentHandler, which only exists once the lifespan has run, while subscribers (often on routers) are declared at build time. The README quick start passed no listener and called it "optional".

So services ran without one. Every rebalance (e.g. each rolling deploy) skipped the flush, finished work on revoked partitions was redelivered to the new owner, and late commits for those partitions failed with IllegalStateError (see #86).

Change

Automatic attachment. initialize_concurrent_processing now attaches a listener to every subscriber this library processes concurrently:

• found through the FastStream application in the context (context.get("app").brokers → broker.subscribers, which includes subscribers from included routers); the app registers itself in broker.context for both FastStream and AsgiFastStream;
• selected like the middleware routes messages: Kafka LogicSubscriber, AckPolicy.MANUAL, not batch=True, subscribed by topic or pattern (manually assigned partitions never rebalance);
• the listener resolves the running handler from the context on each revocation, so it survives stop/start and is a no-op while processing is not running;
• a user's own listener= is kept and called after the flush (sync or async, as aiokafka allows); an explicit ConsumerRebalanceListener is left untouched; a second initialize does not wrap twice.

New parameter rebalance_flush_timeout_sec (default 10 s) configures the attached listener's flush budget.

Logged at ERROR, once per initialize, when it cannot attach: the broker has already started (the listener is fixed at consumer.subscribe()), or the context holds no FastStream application. Both leave a subscriber that needs the flush without one, so every rebalance redelivers finished work until the setup is fixed; that is a misconfiguration an error reporter should surface. Subscribers that do not need a listener are skipped silently.

Explicit fallback. ConsumerRebalanceListener.from_context(context, flush_timeout_sec=...) builds the same context-resolving listener for setups where attachment is not possible. KafkaConcurrentHandler.committer is a new read-only property it uses. The existing constructor and handler.create_rebalance_listener() are unchanged.

The private attribute

FastStream has no public way to add a rebalance listener after a subscriber is declared, so attachment writes subscriber._listener, which FastStream reads when the consumer subscribes. The coupling is bounded:

• the FastStream dependency is pinned <0.8, and the floors CI leg runs 0.7.1;
• a guard test fails if LogicSubscriber.start or ConcurrentBetweenPartitionsSubscriber.start stops passing listener=self._listener;
• the integration test below exercises the real consumer.

Setting a listener after declaration is not possible through any public FastStream API, and the alternative (every subscriber passing listener=) is what services were not doing.

Tests

• Unit (real KafkaBroker / KafkaRouter / FastStream, no network; failed before the change): attached to MANUAL subscribers on the broker and on an included router with the configured timeout; attached listener flushes the running handler; not attached to non-MANUAL, batch, or partition-assigned subscribers; user listener kept and called (sync and async); explicit listener left alone; no double wrap after restart; ERROR when no app; ERROR naming the topic when the broker is already running; FastStream guard test.
• from_context unit tests: handler registered after the listener is resolved; no handler → no-op; stopped handler → no flush.
• Integration against real Kafka: broker 1 (in a FastStream app, no listener=) processes a message with commit batch size/timeout out of reach, then a second member joins the group; the offset must be committed by the revoke callback. With the attach_rebalance_listeners call removed the test fails ({} == {0: 1}).
• Full suite (just test): 211 passed, 100% coverage. Unit suite also passes on the floors (FastStream 0.7.1, aiokafka 0.14.0). ruff and ty clean.