Skip to content

Topic Configuration#

By default, FastStream creates every topic your subscribers consume from with a single partition and a replication factor of 1. To configure a topic, pass a Topic object instead of a plain topic name:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
from faststream import FastStream, Logger
from faststream.kafka import KafkaBroker, Topic

broker = KafkaBroker("localhost:9092")
app = FastStream(broker)


@broker.subscriber(
    Topic("orders", num_partitions=3, replication_factor=2),
    Topic("legacy-orders", declare=False),
    "audit",
)
async def on_order(msg: str, logger: Logger):
    logger.info(msg)

Topic and plain strings can be mixed freely — a string is just a shortcut for Topic(name), so "audit" above is created with the defaults.

Options#

Option Default Description
num_partitions 1 Number of partitions to create the topic with.
replication_factor 1 Replication factor to create the topic with.
declare True Whether FastStream creates the topic for you.

Settings apply at creation time only: Kafka ignores them for a topic that already exists, so changing num_partitions will not repartition a live topic.

Opting a Topic out of Creation#

Set declare=False for topics that somebody else provisions — another service, or your infrastructure-as-code:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
from faststream import FastStream, Logger
from faststream.kafka import KafkaBroker, Topic

broker = KafkaBroker("localhost:9092")
app = FastStream(broker)


@broker.subscriber(
    Topic("orders", num_partitions=3, replication_factor=2),
    Topic("legacy-orders", declare=False),
    "audit",
)
async def on_order(msg: str, logger: Logger):
    logger.info(msg)

FastStream then simply skips the creation request for that topic. It does not check whether the topic exists and does not fail if it is missing, so your consumer starts either way.

Note

declare=False narrows creation down for a single topic. To turn topic creation off for the whole broker, use KafkaBroker(allow_auto_create_topics=False) — that flag always wins, and no topic is created regardless of its declare value.

Unlike faststream.confluent, this flag is FastStream-side only. aiokafka has no allow.auto.create.topics consumer option, so the cluster's own auto.create.topics.enable is a separate switch and still applies to produce/metadata fetches.

consumer_only=True also skips creation (no admin client is started). Topics must already exist.

Partition-assigned subscribers#

A TopicPartition carries the same declare flag. Set it to False when the assigned topic is provisioned elsewhere:

from faststream.kafka import KafkaBroker, TopicPartition

broker = KafkaBroker()

@broker.subscriber(partitions=[TopicPartition("orders", 0, declare=False)])
async def on_order(msg: str) -> None: ...

Publishers#

@broker.publisher(...) accepts a Topic too, for symmetry with subscribers. FastStream never creates publisher topics, though, so only the topic name is used and the creation settings are ignored.