warren.pubsub.kafka.aiokafka
aiokafka-based implementation of the Kafka pubsub layer.
The protocol-level configs in config describe what
to set up; the modules here use aiokafka to actually do it.
aiokafka is an optional dependency: importing this sub-package
without it installed raises
OptionalDependencyError pointing at the
warren[kafka] extra.
Classes
KafkaConnectionManager
classKafkaConnectionManager(config, *, name=None)paramconfigKafkaConnectionConfigparamnamestr | None= NoneattributeadminAIOKafkaAdminClientThe started admin client, for the topology helpers.
methodsetupasync () -> NoneStart the admin client and verify the cluster is reachable.
Builds the ssl.SSLContext from the configured cert paths when
security_protocol == "SSL". Fails loud when the cluster
cannot be reached — the boot-time behavior of
aio_pika.connect_robust for the RabbitMQ backend.
methodteardownasync () -> Nonemethodcreate_consumer(*topics, **kwargs) -> AIOKafkaConsumerCreate an (unstarted) consumer bound to the configured cluster.
paramtopicsstrTopics to subscribe to.
paramkwargsAnyPassed through to AIOKafkaConsumer (group id, offset management, etc. — the consumer manager's concerns).
methodcreate_producer(**kwargs) -> AIOKafkaProducerCreate an (unstarted) producer bound to the configured cluster.
paramkwargsAnyPassed through to AIOKafkaProducer (acknowledgment and idempotence settings — the publisher's concerns).
KafkaConsumerManager
classKafkaConsumerManager(config, connection_manager, consumer, *, data_publisher=None, control_publisher=None, observer_publisher=None, retry_config=None, extract_identity_func=None, publish_hard_failures=True)Kafka consumer manager that bridges message queue operations with application workers.
This class handles all Kafka-specific concerns (consumer lifecycle, offset management, retry logic) while delegating business logic to a worker function that remains agnostic of the underlying message transport.
The consumer receives plain dictionaries and returns optional response dictionaries—it never interacts with Kafka directly.
Offset semantics — the Kafka mapping of RabbitMQ's ack/nack/reject:
- commit after processing ≙ ack / reject(requeue=False)
- seek back without committing ≙ nack(requeue=True)
Messages are processed sequentially per consumer (one in flight at a
time) — the faithful mapping of prefetch_count: 1 — which keeps
offset management trivially correct.
paramconfigKafkaConsumerManagerConfigparamconnection_managerKafkaConnectionManagerparamconsumerMessageConsumerInterfaceparamdata_publisherPublisherInterface | None= Noneparamcontrol_publisherPublisherInterface | None= Noneparamobserver_publisherPublisherInterface | None= Noneparamretry_configRetryConfig | None= Noneparamextract_identity_funcExtractMessageIdentityFunc | None= Noneparampublish_hard_failuresbool= Truemethodsetupasync () -> NoneInitialize Kafka resources for consumption.
- Sets up all publishers.
- Ensures the configured topic exists.
- Creates and starts the Kafka consumer with manual offset commits (the ack/nack seam).
- Must be called before start_consuming().
methodstart_consumingasync () -> NoneBegin consuming messages from the configured topic.
Launches the poll loop: messages are fetched one at a time and processed sequentially (≙ prefetch_count 1), with the offset committed only after processing resolves.
methodstop_consumingasync () -> NoneGracefully stop message consumption and close Kafka resources.
- Signals shutdown (stops accepting new work)
- Cancels the poll loop (no new messages fetched)
- Waits for the in-flight message to complete (with timeout); when it completes it commits per the normal decision matrix
- Stops the Kafka consumer and tears down publishers
Any offset left uncommitted is redelivered to the group after the rebalance — at-least-once, matching nack-on-shutdown on RabbitMQ.
On drain timeout: an async worker is awaited and its task simply outlives the timeout, but a sync worker dispatched via the executor runs in a thread that cannot be force-cancelled — the thread keeps running after we give up waiting. Either way the offset stays uncommitted, so the message redelivers to the group after the rebalance (at-least-once). We do not attempt to kill the worker; we only stop waiting for it.
methodhealthasync (*, probe_timeout=1.0) -> ConsumerHealthKafka has no channel and no blocked state; liveness is the poll loop.
paramprobe_timeoutfloat= 1.0KafkaPublisher
classKafkaPublisher(connection_manager, topic_config, *, route=None, route_func=None, content_type='application/json', name=None)Kafka publisher for a single topic.
A topic plays the role of a fanout exchange in the Kafka backend, so
the publisher is fanout-only: routes are rejected with ValueError
(the parity of RMQPublisher's fanout mode). Messages are sent
without a key, so the cluster distributes them round-robin over the
topic's partitions.
paramconnection_managerKafkaConnectionManagerparamtopic_configKafkaTopicConfigparamrouteRoute | None= Noneparamroute_funcRouteFunc | None= Noneparamcontent_typestr= 'application/json'paramnamestr | None= Nonemethodsetupasync () -> NoneEnsure the publisher's topic exists and start the producer.
The producer is idempotent and waits for full-ISR acknowledgment
(acks="all") — the durability counterpart of RabbitMQ's
persistent delivery mode.
On partial failure the caller is responsible for cleanup via
teardown() (idempotent, best-effort).
Raises:
PubSubSetupError: if the topic or producer cannot be set up.
methodteardownasync () -> Nonemethod__call__async (message) -> Noneparammessagedict[str, Any]Functions
ensure_topic
functionensure_topicasync (connection_manager, config) -> NoneEnsure the topic described by config exists on the cluster.
If the topic is missing and config.create_if_missing is enabled,
it is created — idempotently, so concurrent creation by another
worker is tolerated. If it is missing and creation is disabled
(platforms where topics are provisioned out-of-band), setup fails
loud.
paramconnection_managerKafkaConnectionManagerparamconfigKafkaTopicConfigRaises:
PubSubSetupError: if the topic is missing whilecreate_if_missingis disabled, or if the existence check or the creation fails.