🚀 New issue to ag2ai/faststream by @dzeveloper
📝 Feature: Consumer-Only Mode for KafkaBroker (#2879)


When deploying a subscriber-only service, it's common to provision Kafka credentials scoped strictly to READ + DESCRIBE ACLs on specific topics. Today, KafkaBroker always starts a producer during broker.start(), regardless of whether any @broker.publisher is registered. This causes the producer to attempt a connection using consumer-only credentials, which fails or raises errors at the broker/ACL level.

Proposal:

# Option A: broker-level flag
broker = KafkaBroker("localhost:9092", consumer_only=True)

# Option B: skip producer if no publishers are registered (automatic)
# No API change needed — broker inspects registered publishers at start()

#enhancement #good_first_issue #confluent #aiokafka #faststream #ag2ai
sent via relator

🚀 New issue to ag2ai/faststream by @kumaranvpl
📝 Ability to configure topic params in confluent create_topics (#1827)


Current state: every topic FastStream creates is created with num_partitions=1, replication_factor=1, hardcoded:

faststream/faststream/confluent/helpers/admin.py

Lines 52 to 55 in 2e36197

The only knob users have is the broker-wide allow_auto_create_topics flag — all topics of a broker are created, or none of them are. There is no way to configure a single topic, nor to opt a single topic out of creation.

Request: introduce a Topic schema object accepted by subscriber() / publisher() alongside plain strings, carrying per-topic settings:

from faststream.confluent import KafkaBroker, Topic

broker = KafkaBroker()

@broker.subscriber(
    Topic("topic-name", num_partitions=3),
    Topic("topic-name2", num_partitions=1, replication_factor=2),
    Topic("externally-managed", declare=False),
    "topic-without-settings",
)
async def handler(msg: str) -> None: ...

Scope

1.

 num_partitions / replication_factor per topic — the original ask, from https://github.com/airtai/faststream/discussions/1821. Today both are pinned to 1, which makes FastStream-created topics unusable in any real deployment.

2.

 declare: bool = True — opt a single topic out of creation while auto-creation stays the default for everything else. This subsumes #2679: the global default is not changing, because a broker-wide switch is the wrong granularity — a service typically owns some of its topics and consumes others that are provisioned by a different team or by IaC.

Proposed semantics of declare=False: skip the create_topics call for that topic and nothing else — do not probe the cluster for existence, do not fail if the topic is missing. This matches what allow_auto_create_topics=False does today (a warning, then let the consumer proceed) and matches NATS' JStream(declare=False). Note this deliberately differs from RabbitMQ's RabbitQueue(declare=False), which maps to AMQP passive=True and does raise when the queue is absent — Kafka has no cheap equivalent of a passive declare.

Interaction with the broker-level flag: allow_auto_create_topics=False on the broker keeps winning over everything — it stays the "create nothing at all" switch. declare only narrows creation further when the broker-level flag is on.

Consistency across brokers: declare is the established name for this in FastStream — RabbitQueue(declare=...), JStream(declare=...), KvWatch(declare=...), ObjWatch(declare=...). Topic should use the same name rather than inventing a Kafka-specific one.

AioKafka is out of scope. FastStream never creates topics for faststream.kafka — there is no AdminClient.create_topics call on that path, and aiokafka (0.13.0) does not support an allow_auto_create_topics consumer option at all:

$ grep -rn "auto_create\|auto\.create" .venv/lib/python3.11/site-packages/aiokafka/
$ grep -rn "auto_create" faststream/kafka/

Both return nothing. Topic creation on that path is entirely the Kafka server's auto.create.topics.enable, which FastStream cannot influence. Topic may still be accepted there later for symmetry, but declare would be a no-op, so it should not block this issue.

Implementation notes

• create_subscriber() / create_publisher() signatures must accept str | Topic — *topics: str is what currently trips mypy on the branch in progress.
• Normalise str → Topic(name) at registration time, as RabbitMQ does with str → RabbitQueue.
• AsyncConfluentConsumer.topics_to_create should filter on declare —

 faststream/faststream/confluent/helpers/client.py

 Lines 292 to 294 in 2e36197

• Topic needs __hash__ / __eq__ consistent with each other, since topics end up as dict keys (see #2796 for the RabbitMQ precedent).

Related: #2679 (closed in favour of this), #1486, #1658, #2451.


#good_first_issue #confluent #kafka #faststream #ag2ai
sent via relator

🚀 New issue to ag2ai/faststream by @kumaranvpl
📝 Ability to configure topic params in confluent create_topics (#1827)


Current state: every topic FastStream creates is created with num_partitions=1, replication_factor=1, hardcoded:

faststream/faststream/confluent/helpers/admin.py

Lines 52 to 55 in 2e36197

The only knob users have is the broker-wide allow_auto_create_topics flag — all topics of a broker are created, or none of them are. There is no way to configure a single topic, nor to opt a single topic out of creation.

Request: introduce a Topic schema object accepted by subscriber() / publisher() alongside plain strings, carrying per-topic settings:

from faststream.confluent import KafkaBroker, Topic

broker = KafkaBroker()

@broker.subscriber(
    Topic("topic-name", num_partitions=3),
    Topic("topic-name2", num_partitions=1, replication_factor=2),
    Topic("externally-managed", declare=False),
    "topic-without-settings",
)
async def handler(msg: str) -> None: ...

Scope

1.

 num_partitions / replication_factor per topic — the original ask, from https://github.com/airtai/faststream/discussions/1821. Today both are pinned to 1, which makes FastStream-created topics unusable in any real deployment.

2.

 declare: bool = True — opt a single topic out of creation while auto-creation stays the default for everything else. This subsumes #2679: the global default is not changing, because a broker-wide switch is the wrong granularity — a service typically owns some of its topics and consumes others that are provisioned by a different team or by IaC.

Proposed semantics of declare=False: skip the create_topics call for that topic and nothing else — do not probe the cluster for existence, do not fail if the topic is missing. This matches what allow_auto_create_topics=False does today (a warning, then let the consumer proceed) and matches NATS' JStream(declare=False). Note this deliberately differs from RabbitMQ's RabbitQueue(declare=False), which maps to AMQP passive=True and does raise when the queue is absent — Kafka has no cheap equivalent of a passive declare.

Interaction with the broker-level flag: allow_auto_create_topics=False on the broker keeps winning over everything — it stays the "create nothing at all" switch. declare only narrows creation further when the broker-level flag is on.

Consistency across brokers: declare is the established name for this in FastStream — RabbitQueue(declare=...), JStream(declare=...), KvWatch(declare=...), ObjWatch(declare=...). Topic should use the same name rather than inventing a Kafka-specific one.

AioKafka is out of scope. FastStream never creates topics for faststream.kafka — there is no AdminClient.create_topics call on that path, and aiokafka (0.13.0) does not support an allow_auto_create_topics consumer option at all:

$ grep -rn "auto_create\|auto\.create" .venv/lib/python3.11/site-packages/aiokafka/
$ grep -rn "auto_create" faststream/kafka/

Both return nothing. Topic creation on that path is entirely the Kafka server's auto.create.topics.enable, which FastStream cannot influence. Topic may still be accepted there later for symmetry, but declare would be a no-op, so it should not block this issue.

Implementation notes

• create_subscriber() / create_publisher() signatures must accept str | Topic — *topics: str is what currently trips mypy on the branch in progress.
• Normalise str → Topic(name) at registration time, as RabbitMQ does with str → RabbitQueue.
• AsyncConfluentConsumer.topics_to_create should filter on declare —

 faststream/faststream/confluent/helpers/client.py

 Lines 292 to 294 in 2e36197

• Topic needs __hash__ / __eq__ consistent with each other, since topics end up as dict keys (see #2796 for the RabbitMQ precedent).

Related: #2679 (closed in favour of this), #1486, #1658, #2451.


#enhancement #good_first_issue #confluent #kafka #faststream #ag2ai
sent via relator

🚀 New issue to ag2ai/faststream by @davorrunje
📝 Docs: dealing with different schema registries (#1297)


It would be good to add documentation with examples how to deal with different schema registries. Again there are many registries and coupling router with a particular registry isn't a good idea, unless there will be a some Abstract class first, so later community can add implementation


#documentation #enhancement #good_first_issue #confluent #faststream #ag2ai
sent via relator