📝 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