Kafka Quick Reference

import jsonfrom confluent_kafka import Producer
producer = Producer({    "bootstrap.servers": "localhost:9092",    "enable.idempotence": True,    "acks": "all",    "linger.ms": 10,})

def on_delivery(err, msg):    if err is not None:        print(f"delivery failed: {err}")

for event in events:    producer.produce(        topic="order-events",        key=str(event["customer_id"]).encode("utf-8"),        value=json.dumps(event).encode("utf-8"),        on_delivery=on_delivery,    )    producer.poll(0)
remaining = producer.flush(timeout=10)

import jsonfrom confluent_kafka import Consumer
consumer = Consumer({    "bootstrap.servers": "localhost:9092",    "group.id": "warehouse-sink",    "auto.offset.reset": "earliest",    "enable.auto.commit": False,})consumer.subscribe(["order-events"])
batch = []try:    while True:        msg = consumer.poll(timeout=1.0)        if msg is None:            continue        if msg.error():            continue
        try:            event = json.loads(msg.value().decode("utf-8"))        except ValueError:            continue  # route to a dead-letter destination instead
        batch.append(event)        if len(batch) >= 100:            write_batch(batch)                    # write first            consumer.commit(asynchronous=False)   # then commit            batch.clear()except KeyboardInterrupt:    passfinally:    if batch:        write_batch(batch)        consumer.commit(asynchronous=False)    consumer.close()

kafka-topics --bootstrap-server localhost:29092 --listkafka-topics --bootstrap-server localhost:29092 \  --create --topic order-events --partitions 3 --replication-factor 1kafka-topics --bootstrap-server localhost:29092 --describe --topic order-eventskafka-topics --bootstrap-server localhost:29092 --alter --topic order-events --partitions 6

kafka-configs --bootstrap-server localhost:29092 \  --alter --entity-type topics --entity-name order-events \  --add-config retention.ms=259200000
kafka-configs --bootstrap-server localhost:29092 \  --describe --entity-type topics --entity-name order-events

kafka-consumer-groups --bootstrap-server localhost:29092 --listkafka-consumer-groups --bootstrap-server localhost:29092 --describe --group warehouse-sinkkafka-consumer-groups --bootstrap-server localhost:29092 \  --group warehouse-sink --topic order-events \  --reset-offsets --to-earliest --execute

kafka-console-producer --bootstrap-server localhost:29092 --topic order-eventskafka-console-consumer --bootstrap-server localhost:29092 --topic order-events --from-beginning

{  "order_id": 101,            // business key: unique, used for de-duplication  "customer_id": 1,           // partition key: keeps a customer's events ordered  "created_at": "2026-01-04T09:12:00",  "amount": 1200.0,  "schema_version": 1         // bump on an incompatible change}