Building Event-Driven Microservices with Python and Kafka

22 October 2026

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.

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.

uv init services/producer --no-package

Run the main.py file in the producer directory to verify the current step.

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.

uv add confluent-kafka faker structlog

Now let's create the producer. In the main.py file, we will add the EventProducer class.

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.

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.

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.

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.

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.

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.

{"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.

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"]

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.

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.

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.

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.

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.

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.

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.

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