API reference¶

Runtime limits and librdkafka client options.

The runtime owns the client settings that its guarantees depend on, e.g. isolation.level and enable.auto.commit, and the options cannot override them. Callbacks in the options, e.g. error_cb, run on the runtime's client threads, so they must be thread-safe functions, not coroutines.

The producer's transactional.id is {transactional_id_prefix}-{random}, new each run. With static membership, i.e. a group.instance.id in the options, it derives from the group and instance IDs instead. A restarted instance then aborts the transaction that its crashed run left open, so read_committed readers do not wait transaction_timeout_ms for it. A second live process with the same instance ID fences the first one's producer.

Parameters:

Raises:

client_config() -> dict[str, Any]

Bases: PotikError

Raised when an offset policy refuses a partition, e.g. resume_or_fail, or returns an invalid offset.

A valid offset is the committed one, or within the partition's log. The error fails the run.

Bases: PotikError, PermanentError

Raised when a handler returns something other than a list or tuple of ProducerRecords.

A core handler that returns a single record raises it, for example. It comes from the handler call, so exception rules see it like any other handler exception. To stop the partition instead of letting a catch-all rule handle it, put on(InvalidOutputError, reraise()) before that rule.

Bases: PotikError

Raised when the consumer loses partitions without a revocation, so another member may already own them.

It happens e.g. when the consumer misses its heartbeats. The error fails the run, which commits nothing more.

Bases: Exception

Marks a failure that would repeat if the handler ran again, e.g. a value that does not decode.

retry() and retry_forever() rules pass it to the next rule instead of retrying. Subclass it for such failures of your own. It does not derive from PotikError, so except PotikError does not catch your subclasses.

Bases: Exception

The base of the errors that potik raises.

Errors of the Kafka client stay KafkaExceptions, and invalid arguments raise ValueError or TypeError.

The base class of middleware, which wraps the stages of a Runtime or a Producer.

Subclass it and override any of these stages:

• lifespan surrounds a run, or the time a Producer is open.
• handle wraps each handler call of a run.
• commit wraps each transaction of a run.
• send wraps each Producer.send().

Each stage except lifespan receives call_next, the rest of the chain. A middleware can run code around it, observe its outcome, and scope state to it, e.g. a span. One class can override every stage, so that e.g. one tracing middleware serves runtimes and producers. A runtime or producer runs its middleware outermost first.

An instance serves one run or one Producer, once, so its state lasts one run. Using it a second time raises RuntimeError. Create the instances where you create the runtime or producer, and pass resources, e.g. a connection pool, to __init__.

lifespan(scope: RunScope | ProducerScope) -> AbstractAsyncContextManager[None]

handle(record: ConsumerRecord, call_next: RecordHandler) -> Sequence[ProducerRecord]

commit(transaction: Transaction, call_next: Callable[[Transaction], Awaitable[None]]) -> None

send(record: ProducerRecord, call_next: RecordSender) -> Delivery

Describes a Producer to a middleware's lifespan.

Attributes:

Describes a run to a middleware's lifespan.

Attributes:

What an offset policy decides on: a newly assigned partition, its committed offset, and its log's bounds.

Parameters:

Attributes:

earliest: int

latest: int

A Kafka producer that sends records at once, outside any transaction.

Use it where there is no input record to commit with, e.g. in an API service, or from a handler. Open it once with async with, e.g. in an API service's lifespan, or around a run, and pass it to handlers through potik.build_runtime(context=...). Exiting waits up to close_timeout for every send to be delivered or fail, then fails the rest. On a normal exit, it then raises the first failure of a send that nobody awaited, e.g. a cancelled one. A producer opens only once.

A send from a handler is at-least-once, unlike the handler's outputs. A retried or replayed handler sends again, and readers see the record before the input record commits.

The producer is idempotent and waits for all in-sync replicas (acks=all). After a fatal error, e.g. a broken idempotent sequence, every later send fails. A callback in options that raises does the same, and also fails the sends that wait for a delivery report. Check failure in a liveness probe to restart the process then.

Parameters:

Raises:

failure: KafkaException | None

from_config(config: Config, *, middleware: Iterable[BaseMiddleware] = (), close_timeout: float = _CLOSE_TIMEOUT) -> Self

send(record: ProducerRecord) -> Delivery

The reply that a ReplyListener.expect() block awaits.

wait() -> ConsumerRecord

Hands each reply from a reply topic to the request in this process that awaits it.

Every process that sends requests, e.g. each replica of an API, reads every partition of the reply topic itself. The consumer belongs to no group and commits nothing. It reads from the end of each partition, as it is when the listener opens, so it never reads a reply to a request from an earlier process. It reads with read_committed, so a transactional responder's reply arrives once its request's offset commits.

The listener does not send requests, and potik sets no convention for correlation IDs. The caller puts each request's ID wherever its responder reads it, e.g. a header or a field of the value, and the responder copies it to its reply, where correlate reads it.

A reply that arrives while nothing awaits it is dropped, e.g. after its request timed out, or one to another process. A request that times out may still be handled. A responder that runs it again sends another reply, and only the first one counts. A reply whose correlation ID correlate cannot read is logged and dropped.

Open it once with async with, e.g. in an API's lifespan, before the API takes requests. Once it opens, it reads every reply to a request that the process sends. Exiting fails the requests that still await a reply. After a fatal consumer error, every awaited and later request fails. Check failure in a liveness probe to restart the process then.

Parameters:

Raises:

failure: KafkaException | None

expect(correlation_id: str | None = None) -> Generator[PendingReply]

Consumes topics, calls a handler for each record, and commits its outputs with the record's offset.

Each transaction produces a batch of outputs and commits their input records' offsets atomically, so each record's outputs appear exactly once to read_committed readers. A run uses one consumer, with the consumer group protocol of Kafka 4.0+, and one transactional producer. Each assigned partition handles its records one at a time, in offset order, unless its topic runs by key (see key_concurrency), and partitions run concurrently. The consumer keeps polling while handlers run, so a slow handler does not make it leave the group.

On a revocation or shutdown, the runtime lets in-flight handlers finish within a grace period, then cancels them, and commits the handled records before it gives up the partitions. A failure of one partition stops it (see Config.stop_failed_partitions). Any other failure, e.g. a failed transaction or lost partitions, stops the run without committing more, and the uncommitted records run again on the next run.

Cancelling run() requests a graceful shutdown and waits for it. If the shutdown fails, run() raises its error instead of the cancellation. The runtime never abandons a blocking client call, so the process supervisor must enforce a hard deadline.

A runtime runs once. Make a new one, with new middleware, for another run. potik.build_runtime() builds one from routers of typed handlers.

Parameters:

Raises:

status() -> RuntimeStatus

liveness(*, slack: float = 30.0) -> str | None

readiness() -> str | None

run(*, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORED, handle_signals: bool = True) -> None

Bases: NamedTuple

Where the broker appended a sent record.

A snapshot of one assigned partition.

Attributes:

Bases: NamedTuple, Generic[_V_co]

A record to produce.

A handler returns it as an output, which commits in one transaction with the input record's offset. Producer.send() sends it at once instead.

ProducerRecord[V] marks a record that a Topic[V] made, for type checkers only. A record made directly is a ProducerRecord[Any].

Attributes:

A snapshot of a run, from Runtime.status().

Times are time.monotonic() values. potik.core.health checks a snapshot for liveness and readiness probes.

Attributes:

The outputs and offsets of a batch of handled records, which commit atomically.

Middleware sees it in the commit stage (see BaseMiddleware.commit).

Attributes:

Bases: BaseMiddleware

Limits how many handler calls run at once across partitions, in total and per topic.

A call takes a slot of its topic first, then a shared one, so a call that waits for its topic holds no shared slot. Only the middleware after it runs with a slot, so put it last. Exception rules before it then do not hold a slot during retry backoff. potik.build_runtime() adds one from its own max_concurrency and its handlers', after the exception rules.

Parameters:

Raises:

What an exception rule decides on: the exception, the record, and the attempt.

Attributes:

Bases: BaseMiddleware

Retries or recovers from a handler's exceptions by asking rules in order, like except clauses.

With potik.build_runtime(), pass the rules as exception_rules instead of creating this middleware. Build rules with on(), retry(), retry_forever(), skip(), reraise(), and dead_letter().

The first rule that returns something other than None decides. RetryAfter, e.g. from retry(), calls the handler again after its delay. Records, e.g. from skip() or dead_letter(), commit as the record's outputs in place of the failure. Reraise, from reraise(), lets the exception propagate. An exception that no rule handles propagates too, and stops the partition, or fails the run (see Config.stop_failed_partitions).

Rules see the single exception of an ExceptionGroup that holds only one, so on(TimeoutError, ...) matches a TimeoutError from a task group in the handler. A group of several exceptions stays a group, which on(ExceptionGroup, ...) or a function matches. Either way, an exception that no rule handles propagates as the handler raised it.

A partition handles one record at a time, so a retry holds up its partition until the record succeeds or a later rule handles it. With key concurrency (see Runtime), a retry holds up only its key, though the records after it commit only once it succeeds or a rule handles it. A retry() rule that has used up its attempts returns None, so a later rule handles the exception. So does a retry() rule for a PermanentError, e.g. a DecodeError, which would fail the same way again.

The middleware's output_topics collects its rules' (see DeclaresOutputs), e.g. the dead letter topics of dead_letter() rules, so that the topic check covers them.

With the otel extra, the middleware records each exception that it retries or handles as an exception event on the current span. Inside TracingMiddleware, that span is the record's process span.

Parameters:

Raises:

ExceptionMiddleware(
    on(is_transient, retry(max_attempts=5)),
    on(Exception, dead_letter(dead_letters)),
)

A backoff whose delay grows by a factor after each attempt, up to a cap.

Parameters:

Raises:

A decision to ask no more rules and let the exception propagate, like a bare raise in an except clause.

The exception then stops the partition, or fails the run (see Config.stop_failed_partitions).

A decision to call the handler again after seconds.

Bases: Protocol

A middleware or exception rule with output_topics, the topics that it writes records to.

dead_letter() rules declare their dead letter topic this way, for example. potik.build_runtime() adds the declared topics of its middleware and exception rules to its topic check. They do not restrict where records go.

output_topics: Sequence[str]

Bases: Protocol

Any object with a topic's name, e.g. a Topic. APIs that take topics take one in place of a name.

name: str

Bases: PotikError

Raised by TopicCheckMiddleware when topics are missing, or the client is not authorized to describe them.

Attributes:

Bases: BaseMiddleware

Checks that topics exist before a run or a Producer starts, and fails the start otherwise.

Put it first, so that a missing topic fails the start before other lifespans open resources. potik.build_runtime() puts one first, with its handled topics and every output topic that its handlers, routers, middleware, and exception rules declare, unless check_topics=False.

The client needs permission to describe each topic. A missing or denied topic fails the start with TopicCheckError, and other broker errors, e.g. a timeout, with KafkaException. Cancelling the start, e.g. on SIGTERM, stops the check. It never creates a missing topic.

Parameters:

Raises:

Bases: BaseMiddleware

Traces each record's handling, its outputs to the transaction that commits them, and each Producer send.

It creates these spans:

• process {topic} (consumer), one per handler call. It continues the trace in the record's headers. A record that runs again, e.g. after a crash, gets a second process span in the same trace.
• create {topic} (producer), one per output, a child of its record's process span. The output's headers carry its context. It ends at once, because backends count producer spans as dependency latency, and the transaction's span covers the produce.
• send {topic} (client), one per transaction, as the root of its own trace. A transaction without outputs is named commit {topic}, and the topic is left out of the name when there are several. It links to the sampled create span of each output, up to the SDK's link limit. These links connect a record's trace to the transaction that commits its outputs. Only a committed handling's outputs reach downstream consumers, so when a record runs again, its committed process span is the one whose create spans have a link and downstream children. A record without outputs has no link to its transaction.
• send {topic} (producer), one per Producer.send(), a child of the current span, e.g. a handler's process span or an API's request span. The record's headers carry its context. It lasts until the broker acknowledges the record.

Each span has the attributes that the OpenTelemetry messaging semantic conventions require for it, e.g. messaging.consumer.group.name on a run's spans. optional_attributes picks the optional ones. To add attributes of your own, or to rename a span, use trace.get_current_span() in a handler or in middleware after this one. The current span there is the process span, the transaction's span, or the send span.

Use one instance for each runtime and each producer. Put it first in the runtime's middleware, so that its spans cover the other middleware. potik.build_runtime() runs the exception rules inside its middleware, so one process span covers a record's retries and the outputs of exception rules, such as dead letters. In a core Runtime, put it before ExceptionMiddleware for the same span. A cancellation, e.g. by a revocation, ends the process span with a cancelled event instead of an error.

A transaction's span is the root of its own trace, so the tracer provider's sampler decides on it apart from the records in it. To sample transactions, use transaction_sample_ratio with a sampler that keeps root spans.

Parameters:

Raises:

DEFAULT_OPTIONAL_ATTRIBUTES: frozenset[str] = frozenset({MESSAGING_DESTINATION_PARTITION_ID, MESSAGING_KAFKA_OFFSET, MESSAGING_CLIENT_ID})

Bases: BaseMiddleware

Measures handler calls, transactions, and Producer sends with the OpenTelemetry messaging metrics.

• messaging.client.consumed.messages counts the records that the run passes to its handler, per topic. Its messaging.operation.name is process, the operation that delivers the records to the application.
• messaging.process.duration measures each handler call, per topic, with error.type if it raised.
• messaging.client.operation.duration measures each transaction, including its retries after a rebalance, with error.type if it failed. Its operation is send, or commit if it has no outputs.
• messaging.client.sent.messages counts each transaction's outputs, per topic, with error.type if the transaction failed.
• For each Producer.send(), messaging.client.operation.duration measures the whole call, including any wait for queue room, and messaging.client.sent.messages adds one, per topic.
• potik.partition.stopped counts the stopped partitions, per topic (see Config.stop_failed_partitions). The probes ignore them (see potik.core.health), so alert when it stays above zero. The semantic conventions do not define this metric.

Each metric has the attributes that the conventions require for it, e.g. messaging.consumer.group.name on a run's durations. optional_attributes picks the optional ones. To rename a metric, drop an attribute, or change the buckets, configure a View in the SDK's MeterProvider.

Use one instance for each runtime and each producer. A handler call's duration includes its retries, because potik.build_runtime() runs the exception rules inside its middleware. In a core Runtime, put it before ExceptionMiddleware for the same duration. Cancelled handler calls, transactions, and sends record no duration and no sent count, but a cancelled handler call still counts as consumed.

Parameters:

Raises:

DEFAULT_OPTIONAL_ATTRIBUTES: frozenset[str] = frozenset({MESSAGING_DESTINATION_PARTITION_ID})

A group of topic handlers, which potik.build_runtime() builds a runtime from.

Register handlers with @router.handler(topic). A router needs no runtime, so a module can define one at import, and tests can call its handlers directly. A router has no middleware of its own. To apply an exception rule to some topics only, use on(..., topics=...), and in middleware, check record.topic().

Parameters:

Raises:

handler(topic: str | Topic[Any, Any], *, max_concurrency: int | None = None, key_concurrency: int | None = None) -> Callable[[H], H]

handler(topic: str | Topic[Any, Any], *, max_concurrency: int | None = None, key_concurrency: int | None = None, output_topics: Iterable[OutputTopic[V]]) -> Callable[[Callable[P, Awaitable[HandlerResult[V]]]], Callable[P, Awaitable[HandlerResult[V]]]]

handler(topic: str | Topic[Any, Any], *, max_concurrency: int | None = None, key_concurrency: int | None = None, output_topics: Iterable[OutputTopic[V] | str]) -> Callable[[Callable[P, Awaitable[HandlerResult[V | bytes]]]], Callable[P, Awaitable[HandlerResult[V | bytes]]]]

handler(topic: str | Topic[Any, Any], *, max_concurrency: int | None = None, key_concurrency: int | None = None, output_topics: Iterable[OutputTopic[Any] | str] = ()) -> Callable[[_UserHandler], _UserHandler]

@router.handler(orders, output_topics=[accepted])
async def accept(order: Order, customer: Annotated[str, Key()]) -> ProducerRecord[Accepted]:
    return accepted(Accepted(order.id), key=customer)

Bases: PotikError

Raised when a router cannot register a handler, or potik.build_runtime() cannot build a runtime from routers.

Bases: Generic[_V, _K]

A Kafka topic's name, and the types and serdes of its values and keys.

Declare each topic once, e.g. in a module that its producers and consumers share, so that they all use the same wire format. Pass it to Router.handler() to handle it, and call it to make a ProducerRecord to it. Type checkers infer its types from the given ones, e.g. Topic("events", A | B, serde=json) is a Topic[A | B]. A topic is immutable.

The declaration gets its decoders and encoders from the serdes, so a serde that does not support a type fails here, not on the first record.

A tombstone, a record with a null value, has no value to decode. A handler's value parameter takes its default for it, else None if its type allows None, e.g. Order | None for a Topic[Order]. Otherwise the handler call raises DecodeError.

Parameters:

Raises:

json = PydanticJson()
orders = Topic("orders", Order, serde=json)  # Topic[Order], str keys
by_number = Topic("orders.by_number", Order, key_type=int, serde=json)

decode(data: bytes) -> _V

decode_key(data: bytes) -> _K

Bases: NamedTopic, Protocol

The type of the topics in a handler's output_topics, e.g. a Topic[V].

You rarely name it yourself. Unlike Topic, it is covariant in V, so type checkers accept output_topics=[a, b], with a: Topic[A] and b: Topic[B], as outputs of A | B.

Marks a parameter, Annotated[T, Key()], that takes the record's key.

T is the topic's key type, or bytes for the raw key. A null key takes the parameter's default if it has one, else None if T allows None, else the call fails with DecodeError.

Marks a parameter, Annotated[T, Header(name)], that takes a header's value.

If the record has the header more than once, the parameter takes the last value. A missing or null header takes the parameter's default if it has one, else None if T allows None, else the call fails with DecodeError.

Parameters:

Marks a parameter, Annotated[T, Context()], that takes the run's context.

The context is the object passed as potik.build_runtime(context=...), e.g. a dataclass of the service's clients. potik does not check it against T. potik.build_runtime() raises RegistrationError if a handler or one of its dependencies takes it and no context is given.

Marks a parameter, Annotated[T, Depends(dependency)], that takes what dependency returns.

A dependency is a function, sync or async, a class, or a callable object, which must be hashable. Its parameters take parts of the record, the run's context, or other dependencies, as a handler's parameters do, and registration checks them against the topic in the same way. A call decodes each part of the record once, e.g. the value, for every parameter that takes it with the same decoder, so they all get the same object. Each one still takes its own default for a null part. A sync dependency runs on the event loop, so it must not block. potik does not check what a dependency returns against T.

A dependency that yields, sync or async, gives the parameter the value that it yields, and runs the rest of its body after the handler returns or raises, e.g. to commit or roll back a transaction. Do not decorate it with @contextmanager or @asynccontextmanager: registration rejects such a function, because potik enters dependencies that yield itself.

Each handler call resolves the handler's dependencies anew, in the call's task, one at a time, in parameter order, and each one after its own dependencies. A dependency runs once per call, even if several parameters take it, in the handler and in its dependencies, or take dependencies equal to it, e.g. the same bound method. A retry of the call by an exception rule resolves them again.

After the call, potik resumes the dependencies that yield, last entered first. Each one resumes even if one resumed before it failed:

• If the handler, or a dependency, raised, yield raises that exception in the dependency. A dependency cannot suppress it: potik raises it even if the dependency catches it, so exception rules still handle it. The one exception is a cancellation by the dependency's own cancel scope around yield: with anyio.move_on_after(), a call that runs out of time returns no outputs, so its record commits, unless the call raised another exception, which it then raises. With anyio.fail_after(), it raises TimeoutError. The scope also covers the code after yield of the dependencies that potik entered after it. A dependency that swallows an outer cancellation, e.g. a revocation's, suppresses nothing: the call still raises it.
• If a dependency raises after yield, the call fails with that exception. Exception rules then handle it as a handler failure, and the handler's outputs are not committed.
• If a revocation or shutdown cancels the call, the code after yield runs in the cancelled scope, where each await raises the cancellation again. Shield the awaits that must finish, with a timeout, e.g. with anyio.move_on_after(5, shield=True):.

The code after yield runs before the transaction commits the handler's outputs. So a write that it commits, e.g. to a database, happens at least once, not exactly once.

Parameters:

Raises:

async def transaction(services: Annotated[Services, Context()]) -> AsyncIterator[Transaction]:
    tx = await services.db.begin()
    try:
        yield tx
    except BaseException:
        with anyio.move_on_after(5, shield=True):
            await tx.rollback()
        raise
    await tx.commit()


@router.handler(orders)
async def save(order: Order, tx: Annotated[Transaction, Depends(transaction)]) -> None:
    await tx.execute("INSERT INTO orders VALUES ($1)", order.sku)

Bases: PotikError, PermanentError

Raised from a handler call when a parameter, of the handler or of a dependency, cannot take its part of the record.

The part fails to decode, or it is missing or null and the parameter has no default and does not allow None. It is a PermanentError, so retry() rules do not retry it.

Attributes:

Bases: PotikError, PermanentError

Raised when a topic cannot encode a key or value.

The key or value has the wrong type, or the topic's serde fails on it. It is a PermanentError, so retry() rules do not retry it.

Attributes:

Bases: Protocol

Makes the decoders and encoders of the types it supports.

A Topic calls decoder() and encoder() once for its keys and once for its values, when it is declared, usually at import, so they must not do I/O. The functions they return run synchronously on the event loop for every record, so they must not do I/O either. When potik decodes a handler parameter, a decoder's exception becomes a DecodeError. When a Topic call encodes a record, an encoder's exception becomes an EncodeError. Both are permanent, so retry() rules do not retry them. For a format whose decoding needs I/O, write an AsyncSerde.

decoder(tp: TypeForm[T], context: SerdeContext) -> Decoder[T] | None

encoder(tp: TypeForm[T], context: SerdeContext) -> Encoder[T] | None

Bases: Protocol

Makes the resolvers and encoders of the types it supports, for formats whose decoding needs I/O.

A resolver does the I/O that decoding one record needs, e.g. fetching the writer schema that the record names from a registry, and returns the decoder of that record. A resolver's exceptions reach the exception rules as they are, so retry() rules retry e.g. a timeout. Raise a PermanentError for a failure that would repeat, e.g. an unknown schema. The decoder that the resolver returns, and the encoders, work as a Serde's do: they are synchronous, must not do I/O, and their failures are permanent.

A Topic calls resolver() and encoder() once for its keys and once for its values, when it is declared, so they must not do I/O. An encoder whose format needs I/O, e.g. the registry ID of its writer schema, gets it before the run. The serde learns every topic's keys and values when they are declared, so a method of its own, which the service's main awaits before the run, can register their schemas. potik does not open or close a client that the serde uses, e.g. a registry's. Open and close it around the run, e.g. in the service's main.

resolver(tp: TypeForm[T], context: SerdeContext) -> Resolver[T] | None

encoder(tp: TypeForm[T], context: SerdeContext) -> Encoder[T] | None

The keys or values of one topic, which a serde makes a decoder and an encoder for.

A serde that depends on the topic reads it, e.g. one that names a schema registry subject after the topic, like Confluent's TopicNameStrategy (orders-value). Other serdes ignore it.

Attributes:

Converts str as UTF-8, int and UUID as their text, and bytes as they are.

It is the default serde of topics' keys and values, and the default decoder of Header parameters. A subclass counts as its base, e.g. a StrEnum as its value, and a NewType counts as its supertype. It does not support bool, or Annotated types such as Annotated[str, Field(min_length=3)], rather than ignore their constraints.

An int decodes only from ASCII digits with an optional leading -, the form that it encodes to, so different keys never decode to the same int.

JSON through pydantic, for any type that pydantic's TypeAdapter supports.

Decoding validates the data against the type. Encoding serializes without validating, but fails on a value that pydantic would warn about, e.g. an int in a str field. Pass warnings=... to change that. Install the pydantic extra to use it.

Parameters:

Raises:

JSON through msgspec, for any type that msgspec supports without dec_hook or enc_hook.

Decoding validates the data against the type. Encoding serializes by the value's own type. Install the msgspec extra to use it.

Raises:

An in-process Kafka cluster that runs a whole app in a test, without a broker.

The app's own clients connect to it, so its runtime, middleware, and topic check run as in production. Point the app at the cluster with config(), create its topics, run the app, and wait for its outputs with wait_records().

The cluster has one broker. It creates no topic on its own, so a topic that the test did not create fails the app's start. It commits transactions and hides uncommitted outputs from readers. It does not store the offsets that a transaction commits, and it does not fence stale group members. So a restarted app handles its input again, and only a real broker can test what a run committed.

Enter it with with. Exit it only after the app's run and every other client of the test have closed, because a client that outlives the cluster keeps trying to reconnect. It runs librdkafka's mock cluster, on Linux and macOS only.

Raises:

async def test_accept() -> None:
    with MockCluster() as cluster:
        cluster.create_topic(orders)
        cluster.create_topic(accepted)
        async with Producer(cluster.bootstrap_servers) as producer:
            await producer.send(orders(Order(sku="A"), key="c1"))
        runtime = build(cluster.config(CONFIG))  # your service's potik.build_runtime(...)
        async with anyio.create_task_group() as tasks:
            await tasks.start(partial(runtime.run, handle_signals=False))
            (record,) = await cluster.wait_records(accepted, 1)
            tasks.cancel_scope.cancel()  # shuts the run down gracefully

bootstrap_servers: str

config(config: Config) -> Config

create_topic(topic: NamedTopic | str, partitions: int = 1) -> None

wait_records(topic: NamedTopic | str, count: int, *, timeout: float = 10.0) -> list[ConsumerRecord]