# Building Event-Driven Microservices with Python and Kafka In this blog, we will create 2 microservices in Python. One will produce events to Kafka, and the other will consume them. We will also containerize the services with Docker and run the Kafka broker with docker-compose. Before building something, let's talk about Kafka and Event-Driven architecture. Kafka is an event streaming platform. A producer produces an event to the Kafka server, known as a broker. Events that belong to the same category or type are grouped into a topic, e.g., `user-signup` or `order-placed` events. These 2 topics can be on the same broker, and the producer will publish messages to a topic on the broker. Multiple consumers can subscribe to these topics and process the events. Let's create a new project directory and add a `docker-compose.yaml` file in the root of the project. Add a Kafka service to it to run locally. ```yaml services: kafka: image: apache/kafka:4.1.1 ports: - "9092:9092" environment: - KAFKA_NODE_ID=1 - KAFKA_PROCESS_ROLES=broker,controller - KAFKA_LISTENERS=INTERNAL://:19092,EXTERNAL://:9092,CONTROLLER://:9093 - KAFKA_ADVERTISED_LISTENERS=INTERNAL://kafka:19092,EXTERNAL://localhost:9092 - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT - KAFKA_INTER_BROKER_LISTENER_NAME=INTERNAL - KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER - KAFKA_CONTROLLER_QUORUM_VOTERS=1@kafka:9093 - KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 ``` Run the `docker compose up kafka` command to verify that the Kafka broker is running. This will run the broker on port 9092. Now let's create the producer, which will publish events to this Kafka broker. We will use `uv` to create and run the services. `uv` makes it easier to manage dependencies. I am running the command from the project root. It will be different depending on your current directory. ```sh uv init services/producer --no-package ``` Run the `main.py` file in the producer directory to verify the current step. ```sh cd services/producer uv run main.py ``` We will use the `confluent-kafka` package to publish messages. The `confluent-kafka` package is a wrapper over `librdkafka` library which is written in C. Other 3rd-party libraries, such as `kafka-python` are written natively in Python and don’t need `librdkafka`. We will also install `structlog` for logging and `faker` to generate random data. ```sh uv add confluent-kafka faker structlog ``` Now let's create the producer. In the `main.py` file, we will add the `EventProducer` class. ```python import socket from dataclasses import dataclass from confluent_kafka import Producer @dataclass(frozen=True) class EventProducerConfig: topic: str servers: str class EventProducer: def __init__(self, config: EventProducerConfig) -> None: self.topic = config.topic self.producer = Producer( { "bootstrap.servers": config.servers, "client.id": socket.gethostname(), "acks": "all", } ) def main(): print("Hello from producer!") if __name__ == "__main__": main() ``` Let's add the logger in this file to print error and info logs. ```python import structlog log = structlog.get_logger() structlog.configure( processors=[ structlog.processors.add_log_level, structlog.processors.TimeStamper(fmt="iso"), structlog.processors.JSONRenderer(), ] ) ``` Then we add the `check_connection` method to the `EventProducer` class to verify the connection between the producer and Kafka brokers. ```python class EventProducer: def check_connection(self, timeout: float = 3.0) -> bool: try: cluster_id = self.producer.cluster_id(timeout=timeout) log.info("kafka_connection_verified", cluster_id=cluster_id) return True except KafkaException as e: log.error("kafka_connection_failed", error=str(e)) return False ``` Now, we will add 3 methods to the EventProducer: the `publish`, `close` and `_acked` methods. The `publish` method takes a string key and a JSON value; it then encodes the JSON value and publishes the message to the previously configured Kafka topic. we call `poll(0)` in this function block. This tells the client not to wait for the current message delivery report and to call the callback for previous messages' delivery reports. The value `0` is the timeout value. The `close` method will flush the pending messages from librdkafka’s internal queue to the broker. If we don't flush the messages, then during shutdown the message in the librdkafka’s internal queue will be lost. The `_acked` method is the acknowledgment callback. If there are no errors, then we are logging the message details like offset, partition, etc. For errors, we print an error log, but we can add custom logic or retry based on business requirements. ```python class EventProducer: def __init__(self, config: EventProducerConfig) -> None: self.topic = config.topic self.producer = Producer( { "bootstrap.servers": config.servers, "client.id": socket.gethostname(), "acks": "all", } ) def check_connection(self, timeout: float = 3.0) -> bool: try: cluster_id = self.producer.cluster_id(timeout=timeout) log.info("kafka_connection_verified", cluster_id=cluster_id) return True except KafkaException as e: log.error("kafka_connection_failed", error=str(e)) return False def _acked(self, err, msg): if err is not None: log.error("delivery_failed", topic=msg.topic(), key=msg.key(), error=str(err)) else: log.info("message_delivered", topic=msg.topic(), partition=msg.partition(), offset=msg.offset()) def publish(self, key: str, msg: dict) -> None: value = json.dumps(msg).encode() self.producer.produce(self.topic, key=key, value=value, callback=self._acked) self.producer.poll(0) def close(self) -> None: remaining = self.producer.flush(10) if remaining: log.error("messages_not_delivered", count=remaining) self.producer.close() ``` Now create a function that will generate some random data and publish it. The function will take the producer instance and a shutdown event to exit from the while loop. Here we are creating some dummy JSON messages with `faker`. The `user_id` is a random number between 1 and 10. And we are using the same `user_id` as the key, so all messages belonging to the same user will go in the same partition. Another important part of the message is the timestamp. As the message will be processed later by the consumer, we will not know when the original event was created. The topic will be auto created for now, but ideally in production environment, it should be explicitly created via script and configure the partition, replication factor etc. ```python def publish_test_data(producer: EventProducer, shutdown: Event) -> None: fake = Faker() while not shutdown.is_set(): user_id = fake.random_int(min=1, max=10) msg = { "id": str(uuid.uuid4()), "user_id": user_id, "name": fake.name(), "email": fake.email(), "created_at": datetime.now(timezone.utc).isoformat(), } try: producer.publish(key=str(user_id), msg=msg) except (BufferError, KafkaException) as e: log.error("publish_failed", error=str(e)) shutdown.wait(0.5) log.info("shutting_down") producer.close() ``` In the main function, we create the config and an instance of `EventProducer`. We will read the Kafka broker address from the environment variable. We will also add a `handle_signal` callback function to handle graceful shutdown. ```python def main(): try: config = EventProducerConfig( topic="test-topic", servers=os.environ.get("KAFKA_BROKERS", "localhost:9092"), ) producer = EventProducer(config=config) if not producer.check_connection(timeout=3.0): print( f"\n[Error] Could not connect to Kafka brokers at {config.servers}. Exiting." ) return shutdown = Event() def handle_signal(signum: int, frame: FrameType | None) -> None: shutdown.set() signal.signal(signal.SIGINT, handle_signal) signal.signal(signal.SIGTERM, handle_signal) publish_test_data(producer=producer, shutdown=shutdown) except Exception as e: log.error("unexpected_error", error=str(e)) print(f"\n[Error] An unexpected error occurred: {e}") ``` After this, when we run `uv run main.py`, we will get the logs of messages produced to the Kafka brokers. If not, then we will get an error. ```json {"cluster_id": "5L6g3nShT-eMCtK--X86sw", "event": "kafka_connection_verified", "level": "info", "timestamp": "2026-10-21T11:33:27.225182Z"} {"topic": "test-topic", "partition": 0, "offset": 39, "event": "message_delivered", "level": "info", "timestamp": "2026-10-21T11:33:27.759329Z"} {"topic": "test-topic", "partition": 0, "offset": 40, "event": "message_delivered", "level": "info", "timestamp": "2026-10-21T11:33:28.263488Z"} {"topic": "test-topic", "partition": 0, "offset": 41, "event": "message_delivered", "level": "info", "timestamp": "2026-10-21T11:33:28.765134Z"} ``` Now, we will create a `Dockerfile` to containerize the producer and run it from the `docker-compose.yaml` file. ```dockerfile FROM python:3.14-slim COPY --from=ghcr.io/astral-sh/uv:0.12.23 /uv /usr/local/bin/uv WORKDIR /app ENV UV_COMPILE_BYTECODE=1 ENV UV_LINK_MODE=copy COPY pyproject.toml uv.lock ./ RUN uv sync --frozen --no-dev --no-install-project COPY . . RUN uv sync --frozen --no-dev CMD ["uv", "run", "main.py"] ``` ```yaml services: producer: build: services/producer depends_on: - kafka environment: KAFKA_BROKERS: kafka:19092 ``` We can build the image with the `docker compose build producer` command and run it with the `docker compose up producer` command. After running the container, we will see the logs for publish messages. ```json Attaching to python-kafka-microservices-producer-1 python-kafka-microservices-producer-1 | Bytecode compiled 907 files in 51ms python-kafka-microservices-producer-1 | {"cluster_id": "5L6g3nShT-eMCtK--X86sw", "event": "kafka_connection_verified", "level": "info", "timestamp": "2026-10-21T11:50:37.507142Z"} python-kafka-microservices-producer-1 | {"topic": "test-topic", "partition": 0, "offset": 49, "event": "message_delivered", "level": "info", "timestamp": "2026-10-21T11:50:38.024508Z"} python-kafka-microservices-producer-1 | {"topic": "test-topic", "partition": 0, "offset": 50, "event": "message_delivered", "level": "info", "timestamp": "2026-10-21T11:50:38.529734Z"} ``` Now, we will create a consumer for the messages. From the root of the project, initialize the consumer service. And install the `confluent-kafka` and `structlog` packages in the service. ```sh uv init services/consumer --no-package cd services/consumer uv add confluent-kafka structlog ``` We will create the `EventConsumer` class in the `main.py` file and `EventConsumerConfig` to pass the broker address and consumer group ID, etc. The consumer group ID is needed to identify a group of similar consumers. They run in parallel, share the load and give high availability. ```python import socket from dataclasses import dataclass from confluent_kafka import Consumer @dataclass(frozen=True) class EventConsumerConfig: topic: str group_id: str servers: str class EventConsumer: def __init__(self, config: EventConsumerConfig) -> None: self.topic = config.topic self.consumer = Consumer( { "bootstrap.servers": config.servers, "client.id": socket.gethostname(), "group.id": config.group_id, "auto.offset.reset": "earliest", "enable.auto.commit": False, } ) ``` Let's add the `consume` method to the `EventConsumer` class. It takes 2 arguments: a `stop` event to stop consuming and an `on_message` callback function, which will take the message as an argument. Here we are subscribing to one topic, but a consumer can subscribe to multiple topics, e.g., `user-created`, `user-updated`, etc. Here, we run a while loop that polls messages in Kafka every second, and if there are no messages, we continue. But if there is any message, we are calling the callback function and then committing back to the broker that we have consumed the message. When the stop event is set, it exits the while loop and closes the consumer. ```python class EventConsumer: def consume(self, stop: Event, on_message: Callable[[Message], None]) -> None: self.consumer.subscribe([self.topic]) log.info( "consumer_started", topic=self.topic, ) try: while not stop.is_set(): msg = self.consumer.poll(timeout=1.0) if msg is None: continue if msg.error(): raise KafkaException(msg.error()) else: on_message(msg) self.consumer.commit(message=msg, asynchronous=True) finally: log.info("closing_consumer...") self.consumer.close() log.info("consumer_stopped") ``` Now, we just need to initialize this `EventConsumer` in the main function and call the consume method with some placeholder `on_message` callback. We are also adding a `handle_signal` callback function to gracefully shut down the consumer. ```python def process_msg(msg: Message) -> None: value = msg.value() if value is None: return log.info("processing_message", value=value.decode("utf-8")) def main(): try: config = EventConsumerConfig( topic="test-topic", group_id="test-group", servers=os.environ.get("KAFKA_BROKERS", "localhost:9092"), ) consumer = EventConsumer(config=config) shutdown = Event() def handle_signal(signum: int, frame: FrameType | None) -> None: shutdown.set() signal.signal(signal.SIGINT, handle_signal) signal.signal(signal.SIGTERM, handle_signal) consumer.consume(stop=shutdown, on_message=process_msg) except Exception as e: log.error("unexpected_error", error=str(e)) print(f"\n[Error] An unexpected error occurred: {e}") ``` When we run `uv run main.py`, we will see logs of events consumed by this consumer. We can containerize the consumer with Docker. The Dockerfile will be exactly similar to the producer. Then we can add this consumer service to the `docker-compose.yaml` file. ```yaml consumer: build: services/consumer depends_on: - kafka environment: KAFKA_BROKERS: kafka:19092 ``` We can run the consumer with the `docker compose up consumer` command. It will run the consumer in Docker and start consuming events. We can also update the consumer replicas in docker-compose to run them in high availability. ```yaml consumer: build: services/consumer depends_on: - kafka environment: KAFKA_BROKERS: kafka:19092 deploy: replicas: 3 ``` But when we see the logs, only one consumer is consuming, and the others are idle. This is because we didn't set any partitions for the topic, so it was created with 1 partition. All the messages are going to that partition, even if they have different keys. We need to increase the number of partitions for the topic. I have `kafkactl`; I will use `kafkactl alter topic test-topic --partitions 3` to increase it to 3. After that, when I run the consumers, I can see all consumers consuming the messages. The number of consumer should be less than or equal to partitions. Otherwise extra consumers will be sitting idle. The source code this blog post is available on [Github](https://github.com/rsbh/python-kafka-microservices)