Kafka Streaming skill for Claude Code

A Kafka design is sound when ordering requirements map to message keys, no acknowledged write can be lost to a single broker failure, consumers commit offsets only after processing and tolerate redelivery, poison messages go to a dead-letter topic instead of blocking a partition, and schema changes are checked for compatibility before deploy.

Ground rules

• Ordering exists only within a partition. Choose the key so that everything that must stay ordered shares a key.
• Durability: replication factor 3, min.insync.replicas=2, producer acks=all.
• Assume at-least-once delivery. Make consumers idempotent (dedupe key, upsert, or processed-offset table) rather than relying on exactly-once.
• Commit offsets after processing succeeds, never before.
• Partition count can grow but not shrink, and growing it remaps keys to partitions. Size it up front.
• Offset resets, topic deletion, and retention changes on production topics require user confirmation; always run --dry-run first.

Workflow

- [ ] 1. Inspect cluster version, topic configs, consumer groups, lag
- [ ] 2. Define the event: key, value schema, ordering need, retention
- [ ] 3. Size partitions and set topic configs
- [ ] 4. Producer: idempotent, acks=all, delivery callbacks, batching
- [ ] 5. Consumer: manual commits, idempotent handling, DLQ, rebalance safety
- [ ] 6. Register schema with a compatibility mode
- [ ] 7. Load test; verify lag, ordering, and failure behavior

1. Inspect

BS=broker1:9092
kafka-topics.sh --bootstrap-server $BS --describe --topic orders
kafka-configs.sh --bootstrap-server $BS --describe --entity-type topics --entity-name orders
kafka-consumer-groups.sh --bootstrap-server $BS --describe --group billing    # LAG per partition
kafka-console-consumer.sh --bootstrap-server $BS --topic orders --from-beginning \
  --max-messages 5 --property print.key=true --property print.timestamp=true

kcat -b $BS -t orders -C -c 5 -f '%k %p %o %s\n' is a handy alternative.

2–3. Topic design

Decision	Guidance
Key	Entity whose events must stay ordered (order_id, account_id). Null key = no ordering
Hot keys	One key with most traffic overloads one partition; add a sub-key or salt only if per-key order can be relaxed
Partitions	Max consumer parallelism = partitions. Size for peak throughput / per-consumer throughput, with 2–3x headroom (e.g. 12, 24, 48)
Retention	Time-based for events (retention.ms); cleanup.policy=compact for latest-state-per-key topics
Message size	Keep under ~1 MB (default limit); store large payloads in object storage and send a reference

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

4. Producer (confluent-kafka Python; same configs in other clients)

from confluent_kafka import Producer

producer = Producer({
    "bootstrap.servers": "broker1:9092",
    "acks": "all",
    "enable.idempotence": True,     # no duplicates or reordering from producer retries
    "linger.ms": 10,
    "compression.type": "zstd",
    "delivery.timeout.ms": 120000,
})

def on_delivery(err, msg):
    if err is not None:
        log.error("delivery failed", extra={"key": msg.key(), "error": str(err)})
        # route to retry/outbox; never ignore

producer.produce("orders", key=order_id.encode(), value=payload, on_delivery=on_delivery)
producer.poll(0)                    # serve delivery callbacks
# on shutdown:
producer.flush(30)

Writing to a database and publishing an event in one action: use the transactional outbox pattern (write the event to an outbox table in the same DB transaction; a relay or CDC tool like Debezium publishes it). Dual writes lose events on crashes.

5. Consumer

from confluent_kafka import Consumer, KafkaError

consumer = Consumer({
    "bootstrap.servers": "broker1:9092",
    "group.id": "billing",
    "enable.auto.commit": False,
    "auto.offset.reset": "earliest",
    "partition.assignment.strategy": "cooperative-sticky",
    "max.poll.interval.ms": 300000,
})
consumer.subscribe(["orders"])
try:
    while running:
        msg = consumer.poll(1.0)
        if msg is None:
            continue
        if msg.error():
            if msg.error().code() != KafkaError._PARTITION_EOF:
                log.error("consumer error", extra={"error": str(msg.error())})
            continue
        try:
            handle(msg)                              # idempotent: upsert keyed by event id
        except PermanentError as exc:
            send_to_dlq(msg, reason=str(exc))        # orders.dlq with original headers + error
        consumer.commit(message=msg, asynchronous=False)
finally:
    consumer.close()                                 # leaves the group cleanly, commits nothing extra

Concern	Rule
Delivery semantics	Process then commit = at-least-once. Commit then process = at-most-once (data loss on crash)
Duplicates	Dedupe by event ID in the sink (unique constraint, upsert, or processed-ID table)
Poison message	After N retries, publish to <topic>.dlq with error metadata and move on
Transient failure	Retry with backoff in-process for short outages; pause the partition for long ones
Slow processing	Keep each poll loop under max.poll.interval.ms, or the consumer is kicked out and its partitions rebalance
Rebalance churn	Cooperative-sticky assignor; static membership (group.instance.id) for stable pods; the new KIP-848 protocol (group.protocol=consumer, Kafka 4.0+) where clients support it

Exactly-once: what it really covers

Kafka transactions (transactional.id, consumers with isolation.level=read_committed) give exactly-once for read-from-Kafka, process, write-to-Kafka loops (Kafka Streams processing.guarantee=exactly_once_v2). They do not make side effects in databases, emails, or HTTP calls exactly-once. For those, use idempotent sinks or the outbox pattern.

6. Schema Registry

Compatibility	Allows	Use when
BACKWARD (default)	New schema reads old data: delete fields, add optional fields with defaults	Consumers upgrade before producers
FORWARD	Old schema reads new data: add fields, delete optional fields	Producers upgrade first
FULL	Both	Independent deploys, long retention
*_TRANSITIVE	Checked against all versions, not just the last	Replay from old offsets

Safe changes: add a field with a default; make a field optional. Breaking: rename, change type, remove a required field. Test compatibility in CI against the registry’s compatibility endpoint (or the Maven/Gradle plugin) before deploy.

7. Verify

• Lag: kafka-consumer-groups.sh --describe --group billing stays bounded at peak; alert on lag in time (seconds behind), not just record count.
• Ordering: for a sample key, consumed offsets and business sequence numbers are monotonic.
• Failure drills: kill a consumer mid-batch (messages redelivered, no duplicates in sink); stop one broker (producer continues with acks=all, no acknowledged loss); publish a malformed message (lands in DLQ, partition keeps flowing).

Lag and incident triage

Symptom	Likely cause	Action
Lag grows on all partitions	Consumers too slow or too few	Scale consumers up to partition count; optimize handler; batch sink writes
Lag on one partition only	Hot key or poison message	Inspect that partition’s head message; DLQ it; reconsider key
Constant rebalancing	Poll loop exceeds max.poll.interval.ms, pods restarting	Smaller max.poll.records, faster handler, static membership
Duplicates in sink	Redelivery after crash/rebalance	Idempotent sink
Missing messages	Auto-commit before processing, acks=1, or retention expired before consumption	Fix commit order, acks, retention
Reprocess needed	Bug fixed, replay required	Stop group, --reset-offsets --to-datetime ... --dry-run, review, then --execute (confirm)

Gotchas

• Adding partitions changes which partition a key maps to; per-key order breaks across the change.
• auto.offset.reset only applies when the group has no committed offset; it is not a replay switch.
• KafkaJS is no longer actively maintained; for new Node services prefer Confluent’s JavaScript client or another maintained library.
• ZooKeeper is gone in Kafka 4.x (KRaft only); old --zookeeper CLI flags no longer exist.
• Compacted topics keep the latest value per key but not immediately; consumers still see older values until compaction runs.
• Large consumer groups across many topics make every rebalance stop-the-world on the classic protocol.

Output

**Topic:** orders — 24 partitions, RF 3, min.insync.replicas 2, 7-day retention, zstd
**Key:** order_id (per-order ordering); schema orders-value v4, BACKWARD compatible
**Producer:** idempotent, acks=all, outbox relay from orders DB
**Consumer (billing):** manual commit after upsert by event_id; DLQ orders.dlq after 5 retries
**Drills:** consumer kill -> 0 duplicates in sink; broker stop -> 0 lost acks; poison msg -> DLQ
**Lag at peak (3k msg/s):** < 2 s