Kafka — Testing Scenarios, Load & CI¶

def test_order_per_key_is_preserved(bootstrap_servers, topics, producer, service):
    for order_id in ("ord-1", "ord-2", "ord-3"):
        for amount in (100, 200, 300):
            produce_json(producer, topics["in"], order_id, new_event(order_id, amount))

    messages = consume_until(bootstrap_servers, topics["out"], lambda ms: len(ms) >= 9)

    per_key = defaultdict(list)
    for m in messages:
        per_key[m.key()].append(json.loads(m.value())["total_cents"])
    assert all(amounts == [100, 200, 300] for amounts in per_key.values()), per_key
    assert all(len({m.partition() for m in messages if m.key() == k}) == 1 for k in per_key)

def test_redelivered_event_is_not_invoiced_twice(bootstrap_servers, topics, producer, service):
    event = new_event("ord-7")
    produce_json(producer, topics["in"], "ord-7", event)
    produce_json(producer, topics["in"], "ord-7", event)               # same event_id: simulated redelivery
    produce_json(producer, topics["in"], "ord-7", new_event("ord-7"))  # marker: a later, different event

    messages = consume_until(bootstrap_servers, topics["out"], lambda ms: len(ms) >= 2)

    sources = [json.loads(m.value())["source_event"] for m in messages]
    assert sources.count(event["event_id"]) == 1

# tests/test_redelivery.py
import pytest
from confluent_kafka import Consumer

from tests.kafka_helpers import new_event, produce_json

pytestmark = pytest.mark.integration


def read_one(bootstrap_servers: str, topic: str, group_id: str, commit: bool) -> int:
    consumer = Consumer({
        "bootstrap.servers": bootstrap_servers,
        "group.id": group_id,
        "auto.offset.reset": "earliest",
        "enable.auto.commit": False,
    })
    consumer.subscribe([topic])
    try:
        for _ in range(50):
            msg = consumer.poll(0.2)
            if msg is not None and not msg.error():
                if commit:
                    consumer.commit(message=msg, asynchronous=False)
                return msg.offset()
        raise AssertionError("no message")
    finally:
        consumer.close()


def test_uncommitted_message_is_redelivered(bootstrap_servers, topic_factory, producer, group_id):
    topic = topic_factory("orders", partitions=1)
    produce_json(producer, topic, "ord-1", new_event("ord-1"))

    first = read_one(bootstrap_servers, topic, group_id, commit=False)    # "crash" before commit
    second = read_one(bootstrap_servers, topic, group_id, commit=True)
    assert first == second == 0                                          # at-least-once: same record again

    produce_json(producer, topic, "ord-1", new_event("ord-1"))
    assert read_one(bootstrap_servers, topic, group_id, commit=True) == 1  # committed -> not redelivered

def test_poison_message_goes_to_dlq_and_consumer_keeps_going(bootstrap_servers, topics, producer, service):
    produce_json(producer, topics["in"], "ord-9", b"{not json")
    produce_json(producer, topics["in"], "ord-9", new_event("ord-9"))

    [dead] = consume_until(bootstrap_servers, topics["dlq"], lambda ms: len(ms) >= 1)
    [invoice] = consume_until(bootstrap_servers, topics["out"], lambda ms: len(ms) >= 1)

    headers = dict(dead.headers())
    assert dead.value() == b"{not json"
    assert headers["error"].startswith(b"JSONDecodeError")
    assert json.loads(invoice.value())["order_id"] == "ord-9"

# tests/lag.py
import time

from confluent_kafka import ConsumerGroupTopicPartitions, TopicPartition
from confluent_kafka.admin import AdminClient, OffsetSpec


def consumer_lag(admin: AdminClient, group: str, topic: str) -> dict[int, int]:
    """Lag per partition = log-end offset - committed offset of the group."""
    partitions = [TopicPartition(topic, p) for p in admin.list_topics(topic, timeout=5).topics[topic].partitions]
    committed = admin.list_consumer_group_offsets(
        [ConsumerGroupTopicPartitions(group, partitions)]
    )[group].result().topic_partitions
    ends = {tp.partition: f.result().offset for tp, f in admin.list_offsets(
        {tp: OffsetSpec.latest() for tp in partitions}
    ).items()}
    # offset < 0: the group never committed this partition -> count all of it (fine for fresh test topics)
    return {tp.partition: ends[tp.partition] - max(tp.offset, 0) for tp in committed}


def wait_for_zero_lag(admin: AdminClient, group: str, topic: str, timeout: float = 30.0) -> float:
    """Return how long the group needed to catch up; fail if it did not within `timeout`."""
    start = time.monotonic()
    while True:
        lag = consumer_lag(admin, group, topic)
        if sum(lag.values()) == 0:
            return time.monotonic() - start
        if time.monotonic() - start > timeout:
            raise AssertionError(f"group {group} still lags on {topic}: {lag}")
        time.sleep(0.5)

def test_consumer_drains_a_burst(admin, bootstrap_servers, topics, producer, service, group_id):
    for i in range(500):
        producer.produce(topics["in"], key=f"ord-{i % 20}", value=json.dumps(new_event(f"ord-{i % 20}")).encode())
    assert producer.flush(10) == 0

    seconds = wait_for_zero_lag(admin, group_id, topics["in"], timeout=60)

    print(f"500 events drained in {seconds:.1f}s")
    assert seconds < 30

docker exec kafka /opt/kafka/bin/kafka-producer-perf-test.sh --bootstrap-server localhost:9092 \
  --topic perf --num-records 100000 --record-size 512 --throughput -1 --command-property acks=all
# 100000 records sent, 41893.6 records/sec (20.46 MB/sec), 602.63 ms avg latency, ...

docker exec kafka /opt/kafka/bin/kafka-consumer-perf-test.sh --bootstrap-server localhost:9092 \
  --topic perf --num-records 100000 --group perf-check

# .github/workflows/kafka-tests.yml (fragment)
jobs:
  tests:
    runs-on: ubuntu-latest
    services:
      kafka:
        image: apache/kafka:4.3.1
        ports:
          - 9092:9092
        env:
          KAFKA_NODE_ID: "1"
          KAFKA_PROCESS_ROLES: broker,controller
          KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
          KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
          KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT
          KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
          KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093
          KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: "1"
          KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: "1"
          KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: "1"
          KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: "0"
          KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false"
        options: >-
          --health-cmd "/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092"
          --health-interval 5s --health-timeout 10s --health-retries 20
    env:
      KAFKA_BOOTSTRAP_SERVERS: localhost:9092
    steps:
      - uses: actions/checkout@v4
      - uses: astral-sh/setup-uv@v6
      - name: Unit and contract tests
        run: uv run pytest -m "not integration"
      - name: Integration tests
        run: uv run pytest -m integration --junitxml=report.xml
      - name: Broker logs on failure
        if: failure()
        run: docker logs "${{ job.services.kafka.id }}" | tail -200