10.3Kafka with Testcontainers

Kafka is one of the most common use cases for Testcontainers. Running a real Kafka broker in your test suite eliminates the fragility of Kafka mocks.

# tests/integration/test_kafka_consumer.py
import pytest
from testcontainers.kafka import KafkaContainer
from confluent_kafka import Producer, Consumer
import json
import time

@pytest.fixture(scope="module")
def kafka():
    """Start a Kafka container for the test module."""
    with KafkaContainer("confluentinc/cp-kafka:7.6.0") as kafka:
        yield kafka

def test_order_event_processing(kafka):
    """Test that order events are correctly produced and consumed."""
    bootstrap_servers = kafka.get_bootstrap_server()

    # Produce a test event
    producer = Producer({"bootstrap.servers": bootstrap_servers})
    order_event = {
        "eventType": "OrderCreated",
        "orderId": "ORD-789",
        "timestamp": "2026-02-09T10:00:00Z",
        "payload": {"items": [{"sku": "WIDGET-A", "qty": 1}]}
    }
    producer.produce("orders", json.dumps(order_event).encode())
    producer.flush()

    # Consume and verify
    consumer = Consumer({
        "bootstrap.servers": bootstrap_servers,
        "group.id": "test-group",
        "auto.offset.reset": "earliest",
    })
    consumer.subscribe(["orders"])

    msg = consumer.poll(timeout=10.0)
    assert msg is not None
    assert msg.error() is None
    event = json.loads(msg.value())
    assert event["eventType"] == "OrderCreated"
    assert event["orderId"] == "ORD-789"

    consumer.close()

def test_consumer_handles_malformed_messages(kafka):
    """Test that the consumer gracefully handles non-JSON messages."""
    bootstrap_servers = kafka.get_bootstrap_server()

    producer = Producer({"bootstrap.servers": bootstrap_servers})
    producer.produce("orders", b"this is not json")
    producer.flush()

    consumer = Consumer({
        "bootstrap.servers": bootstrap_servers,
        "group.id": "test-malformed-group",
        "auto.offset.reset": "earliest",
    })
    consumer.subscribe(["orders"])

    msg = consumer.poll(timeout=10.0)
    assert msg is not None

    # The application's consumer should handle this without crashing
    try:
        json.loads(msg.value())
        assert False, "Should have raised ValueError"
    except (json.JSONDecodeError, ValueError):
        pass  # Expected: consumer should log and skip

    consumer.close()

def test_multiple_partitions(kafka):
    """Test that messages are distributed across partitions."""
    bootstrap_servers = kafka.get_bootstrap_server()

    # Create a topic with multiple partitions using admin client
    from confluent_kafka.admin import AdminClient, NewTopic
    admin = AdminClient({"bootstrap.servers": bootstrap_servers})
    topic = NewTopic("multi-partition", num_partitions=3, replication_factor=1)
    admin.create_topics([topic])

    # Produce messages with different keys (keys determine partition)
    producer = Producer({"bootstrap.servers": bootstrap_servers})
    for i in range(30):
        key = f"customer-{i % 3}"  # 3 different keys
        producer.produce(
            "multi-partition",
            key=key.encode(),
            value=json.dumps({"index": i}).encode(),
        )
    producer.flush()

    # Verify messages are consumable
    consumer = Consumer({
        "bootstrap.servers": bootstrap_servers,
        "group.id": "multi-partition-test",
        "auto.offset.reset": "earliest",
    })
    consumer.subscribe(["multi-partition"])

    messages = []
    deadline = time.time() + 15
    while len(messages) < 30 and time.time() < deadline:
        msg = consumer.poll(timeout=1.0)
        if msg and not msg.error():
            messages.append(json.loads(msg.value()))

    assert len(messages) == 30
    consumer.close()