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.confluent 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#

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.confluent 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.

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.