Kafka Streaming skill for Codex

Done well means: every topic has a written key, partition and retention rationale, producers cannot lose acknowledged writes, consumers commit only after processing and tolerate redelivery, schemas evolve under an enforced compatibility rule, and lag is monitored.

Before changing anything

1. Read AGENTS.md and the client config in code (search for bootstrap.servers, KafkaProducer, Consumer(, @KafkaListener). Identify the client library: Java client, confluent-kafka (librdkafka) for Python, or a framework (Spring Kafka, Kafka Streams).
2. Learn the broker version and platform (Apache Kafka 3.x or 4.x, Confluent, MSK, Redpanda). Kafka 4.0 runs only in KRaft mode (no ZooKeeper) and makes the new consumer rebalance protocol (group.protocol=consumer) generally available.
3. Gather facts for the topic in question. CLI calls need network access to the cluster; in a sandbox, ask the user to run them or approve access:

kafka-topics.sh --bootstrap-server $BS --describe --topic orders
kafka-configs.sh --bootstrap-server $BS --entity-type topics --entity-name orders --describe
kafka-consumer-groups.sh --bootstrap-server $BS --describe --group orders-enricher

Topic design

Decision	Default	Reasoning
Key	The entity whose events must stay ordered (order_id, account_id)	Ordering is per partition only
Partitions	Enough for peak consumer parallelism with headroom (e.g. 12, 24, 48)	Adding partitions later remaps keys and breaks per-key ordering
Replication factor	3	Survives one broker loss with min.insync.replicas=2
min.insync.replicas	2 (with RF 3)	With acks=all, writes fail rather than land on one replica
cleanup.policy	delete for events, compact for latest-state-per-key topics	Compacted topics need non-null keys; null values are tombstones
retention.ms	Longest replay window you need, plus a margin	Consumers offline longer than retention lose data

kafka-topics.sh --bootstrap-server $BS --create --topic orders \
  --partitions 24 --replication-factor 3 \
  --config min.insync.replicas=2 --config retention.ms=604800000

Creating, deleting or reconfiguring topics on a shared cluster needs the user’s confirmation. Never delete a topic or reduce retention without explicit approval.

Producers

Kafka clients since 3.0 default to enable.idempotence=true and acks=all; keep them. Idempotence removes duplicates from producer retries within a session; it does not dedupe your application sending the same event twice.

from confluent_kafka import Producer

p = Producer({
    "bootstrap.servers": BS,
    "enable.idempotence": True,
    "acks": "all",
    "linger.ms": 10,
    "compression.type": "zstd",
    "delivery.timeout.ms": 120000,
})

def on_delivery(err, msg):
    if err:
        log.error("delivery failed key=%s err=%s", msg.key(), err)   # alert, do not swallow

p.produce("orders", key=order_id, value=payload, on_delivery=on_delivery)
p.poll(0)          # serve delivery callbacks
...
p.flush(30)        # before exit; check the returned count of undelivered messages

Key every message whose order matters. Null-key messages are spread across partitions.

Consumers

At-least-once is the practical default: process, then commit. Make processing idempotent (upserts keyed on event ID, dedupe table) so redelivery is harmless.

from confluent_kafka import Consumer, KafkaException

c = Consumer({
    "bootstrap.servers": BS,
    "group.id": "orders-enricher",
    "enable.auto.commit": False,
    "auto.offset.reset": "earliest",       # only applies when the group has no committed offset
    "max.poll.interval.ms": 300000,
    "partition.assignment.strategy": "cooperative-sticky",
})
c.subscribe(["orders"])
try:
    while running:
        msg = c.poll(1.0)
        if msg is None:
            continue
        if msg.error():
            raise KafkaException(msg.error())
        try:
            handle(msg)                     # idempotent
        except PermanentError:
            send_to_dlq(msg)                # orders.dlq with error headers
        c.commit(message=msg, asynchronous=False)
finally:
    c.close()                               # leaves the group cleanly, triggers fast rebalance

Problem	Cause	Fix
Duplicates after restarts	Commit happened before processing finished, or processing is not idempotent	Commit after processing; idempotent writes
Lost messages	Auto-commit committed before a crash; acks=1 on producer	Manual commit; acks=all
Rebalance storms	Processing exceeds max.poll.interval.ms	Smaller batches (max.poll.records in Java), faster handler, or hand work to a bounded pool and pause partitions
Lag grows on one partition	Hot key	Better key, or split the hot entity
Lag grows everywhere	Consumers too slow or too few	Scale consumers up to the partition count; profile the handler
Poison message blocks partition	Retrying forever	Bounded retries, then dead-letter topic

Schemas

• Register Avro, Protobuf or JSON Schema in a Schema Registry; producers serialize with the registry client so every message carries a schema ID.
• Compatibility BACKWARD (the usual default) lets new consumers read old data: add fields with defaults, never remove required fields or change types. Use FULL for topics read by many teams.
• Test a schema change against the registry’s compatibility endpoint before deploying producers.

Exactly-once

Pipeline	Approach
Kafka -> Kafka (consume, transform, produce)	Transactions: transactional.id on the producer, send_offsets_to_transaction, consumers of the output use isolation.level=read_committed
Kafka Streams app	processing.guarantee=exactly_once_v2
Kafka -> database or external API	Not covered by Kafka transactions: idempotent upsert keyed on event ID, or store offsets in the same DB transaction as the data

For producer, consumer and transactional config details read references/client-configs.md when tuning throughput, latency, or setting up transactions.

Replaying and resetting offsets

kafka-consumer-groups.sh --bootstrap-server $BS --group orders-enricher --topic orders \
  --reset-offsets --to-datetime 2025-09-01T00:00:00.000 --dry-run

The group must have no active members. Show the dry-run output to the user and get explicit approval before re-running with --execute; a reset reprocesses (or skips) real data.

Verification

• Produce a known set of keyed test messages to a non-production topic and confirm counts, per-key order, and no duplicates downstream after killing a consumer mid-batch.
• Watch lag per partition (kafka-consumer-groups.sh --describe) during a load test.
• Confirm producer delivery callbacks report zero errors and flush() returns 0.

Report template

**Topic:** orders (24 partitions, RF 3, min.insync.replicas 2, retention 7d, key order_id)
**Producer:** idempotent, acks=all, zstd; delivery errors alert via callback
**Consumer:** group orders-enricher, manual commit after upsert, cooperative-sticky, DLQ orders.dlq
**Guarantee:** at-least-once with idempotent upsert on event_id
**Test:** 100k messages, consumer killed twice: 0 lost, 0 duplicate rows; max lag 1,800