Apache Kafka Cheat Sheet

Producer --> [Topic: Partition 0, Partition 1, Partition 2] --> Consumer Group
 
Topic     = Stream/kategori data (kayak tabel di database)
Partition = Unit paralelisme dalam topic (ordered)
Offset    = Posisi consumer dalam partition
Broker    = Server Kafka
Consumer Group = Group consumer yang share load

# Create topic
kafka-topics.sh --create \
  --topic orders \
  --bootstrap-server localhost:9092 \
  --partitions 3 \
  --replication-factor 1
 
# List topics
kafka-topics.sh --list \
  --bootstrap-server localhost:9092
 
# Describe topic (detail partitions, replicas)
kafka-topics.sh --describe \
  --topic orders \
  --bootstrap-server localhost:9092
 
# Delete topic
kafka-topics.sh --delete \
  --topic orders \
  --bootstrap-server localhost:9092
 
# Alter partitions (increase only)
kafka-topics.sh --alter \
  --topic orders \
  --partitions 6 \
  --bootstrap-server localhost:9092

# Produce ke topic (interactive)
kafka-console-producer.sh \
  --topic orders \
  --bootstrap-server localhost:9092
 
# Produce dari file
kafka-console-producer.sh \
  --topic orders \
  --bootstrap-server localhost:9092 < orders.json
 
# Dengan key
kafka-console-producer.sh \
  --topic orders \
  --bootstrap-server localhost:9092 \
  --property "key.separator=:" \
  --property "parse.key=true"
# Input: order-123:{"product": "laptop", "price": 999}
 
# Dengan header
kafka-console-producer.sh \
  --topic orders \
  --bootstrap-server localhost:9092 \
  --property "headers=event_type=created,source=api"

# Consume dari awal topic
kafka-console-consumer.sh \
  --topic orders \
  --from-beginning \
  --bootstrap-server localhost:9092
 
# Consume real-time (baru saja)
kafka-console-consumer.sh \
  --topic orders \
  --bootstrap-server localhost:9092
 
# Dengan key
kafka-console-consumer.sh \
  --topic orders \
  --property "print.key=true" \
  --property "key.separator= - " \
  --from-beginning \
  --bootstrap-server localhost:9092
 
# Consumer group
kafka-console-consumer.sh \
  --topic orders \
  --group order-processors \
  --bootstrap-server localhost:9092
 
# Lihat offsets per partition
kafka-consumer-groups.sh \
  --group order-processors \
  --describe \
  --bootstrap-server localhost:9092

# List consumer groups
kafka-consumer-groups.sh \
  --list \
  --bootstrap-server localhost:9092
 
# Describe group (lag, offset)
kafka-consumer-groups.sh \
  --describe \
  --group order-processors \
  --bootstrap-server localhost:9092
 
# Reset offset ke earliest
kafka-consumer-groups.sh \
  --reset-offsets \
  --group order-processors \
  --topic orders \
  --to-earliest \
  --execute \
  --bootstrap-server localhost:9092
 
# Reset offset ke latest
kafka-consumer-groups.sh \
  --reset-offsets \
  --group order-processors \
  --topic orders \
  --to-latest \
  --execute \
  --bootstrap-server localhost:9092
 
# Shift offset (-2 = mundur 2 pesan)
kafka-consumer-groups.sh \
  --reset-offsets \
  --group order-processors \
  --topic orders \
  --shift-by -2 \
  --execute \
  --bootstrap-server localhost:9092
 
# Reset ke timestamp spesifik
kafka-consumer-groups.sh \
  --reset-offsets \
  --group order-processors \
  --topic orders \
  --to-datetime 2026-06-21T00:00:00.000 \
  --execute \
  --bootstrap-server localhost:9092

from confluent_kafka import Producer
import json
 
conf = {
    'bootstrap.servers': 'localhost:9092',
    'client.id': 'order-service',
}
 
producer = Producer(conf)
 
def delivery_report(err, msg):
    """Callback dipanggil saat Kafka confirm delivery."""
    if err is not None:
        print(f'Delivery failed: {err}')
    else:
        print(f'Delivered to {msg.topic()}[{msg.partition()}]@{msg.offset()}')
 
# Produce message (async)
producer.produce(
    'orders',
    key=str(order_id),
    value=json.dumps({'product': 'laptop', 'price': 999}).encode('utf-8'),
    callback=delivery_report,
)
 
# Flush untuk memastikan semua pesan terkirim
producer.flush()
 
# Produce dengan header
producer.produce(
    'orders',
    key='order-123',
    value=b'{"product": "laptop"}',
    headers=[('event_type', b'created'), ('source', b'api')],
)
 
# Produce synchronous (blocking, gunakan dengan hemat)
producer.produce('orders', value=b'urgent', sync=True)

from confluent_kafka import Consumer
 
conf = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'order-processors',
    'auto.offset.reset': 'earliest',  # atau 'latest'
    'enable.auto.commit': True,
    'auto.commit.interval.ms': 5000,
}
 
consumer = Consumer(conf)
 
# Subscribe ke satu atau multiple topics
consumer.subscribe(['orders', 'payments'])
 
try:
    while True:
        msg = consumer.poll(timeout=1.0)
 
        if msg is None:
            continue
        if msg.error():
            print(f'Consumer error: {msg.error()}')
            continue
 
        # Process message
        key = msg.key().decode('utf-8') if msg.key() else None
        value = msg.value().decode('utf-8')
        topic = msg.topic()
        partition = msg.partition()
        offset = msg.offset()
        headers = dict(msg.headers()) if msg.headers() else {}
 
        print(f'[{topic}][{partition}][{offset}] key={key} value={value}')
 
except KeyboardInterrupt:
    pass
finally:
    consumer.close()

conf = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'order-processors',
    'enable.auto.commit': False,  # Manual commit
}
 
consumer = Consumer(conf)
consumer.subscribe(['orders'])
 
while True:
    msg = consumer.poll(1.0)
    if msg is None:
        continue
    if msg.error():
        continue
 
    try:
        # Process message
        process_order(msg.value())
        # Commit setelah sukses
        consumer.commit(msg)
    except Exception as e:
        print(f'Error processing: {e}')
        # Tidak commit = pesan akan di-process lagi

from confluent_kafka import Consumer, TopicPartition
 
consumer = Consumer(conf)
 
# Manual assignment ke partition spesifik
consumer.assign([
    TopicPartition('orders', 0, offset=0),  # partition 0, start from offset 0
    TopicPartition('orders', 1, offset=5),  # partition 1, start from offset 5
])
 
# Seek ke offset tertentu
consumer.seek(TopicPartition('orders', 0, offset=100))

# Pesan dengan key yang sama selalu ke partition yang sama
# Berguna untuk ordering guarantee per entity
 
producer.produce('orders', key='user-123', value=data)
producer.produce('orders', key='user-123', value=data2)
# Kedua pesan dijamin ke partition yang sama, jadi ordered
 
# Custom partitioner (Python)
class UserPartitioner:
    def __call__(self, key, all_partitions, available):
        # Route berdasarkan user_id hash
        user_id = int(key.decode())
        return all_partitions[user_id % len(all_partitions)]

Jika consumer join/leave group:
1. Kafka trigger rebalance
2. Partitions di-assign ulang ke consumer yang ada
3. Consumer yang dapat partition baru resume dari committed offset
 
Tips:
- Pastikan processing idempotent (bisa di-process ulang saat rebalance)
- Pakai cooperative rebalancing (Kafka 2.4+) untuk minimize disruption

# Kafka default: at-least-once
# Pesan bisa di-process lebih dari sekali kalau consumer crash sebelum commit
 
# Solusi: idempotent processing
processed_ids = set()
 
def process_message(msg):
    msg_id = extract_id(msg)
    if msg_id in processed_ids:
        return  # Skip, udah diproses
    do_work(msg)
    processed_ids.add(msg_id)

# Retention by time (default 7 hari)
kafka-configs.sh --alter \
  --topic orders \
  --add-config retention.ms=604800000 \
  --bootstrap-server localhost:9092
 
# Retention by size (1 GB per partition)
kafka-configs.sh --alter \
  --topic orders \
  --add-config retention.bytes=1073741824 \
  --bootstrap-server localhost:9092
 
# Compact topic (simpan hanya value terakhir per key)
kafka-configs.sh --alter \
  --topic user-state \
  --add-config cleanup.policy=compact \
  --bootstrap-server localhost:9092

from kafka import KafkaProducer, KafkaConsumer
import json
 
# Producer
producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
    key_serializer=lambda k: k.encode('utf-8') if k else None,
    acks='all',  # Tunggu semua replicas confirm
    retries=3,
    max_in_flight_requests_per_connection=1,  # Preserve order saat retry
)
 
# Send
future = producer.send('orders', key='user-123', value={'product': 'book'})
metadata = future.get(timeout=10)  # Wait for confirmation
print(f'Sent to partition {metadata.partition} at offset {metadata.offset}')
 
producer.flush()
producer.close()
 
# Consumer
consumer = KafkaConsumer(
    'orders',
    bootstrap_servers=['localhost:9092'],
    group_id='order-processors',
    auto_offset_reset='earliest',
    enable_auto_commit=False,
    value_deserializer=lambda x: json.loads(x.decode('utf-8')),
    consumer_timeout_ms=10000,  # Stop setelah 10s tanpa pesan
)
 
for message in consumer:
    print(f'{message.topic}:{message.partition}:{message.offset}')
    print(f'  Key: {message.key}')
    print(f'  Value: {message.value}')
    consumer.commit()  # Manual commit

# Simpan semua perubahan state sebagai event stream
# Topic: account-events
# Key: account_id
# Value: {"type": "deposit", "amount": 100, "timestamp": "..."}
 
# Rebuild state dari events
def rebuild_state(consumer, account_id):
    balance = 0
    for msg in consumer:
        if msg.key == account_id:
            event = json.loads(msg.value)
            if event['type'] == 'deposit':
                balance += event['amount']
            elif event['type'] == 'withdraw':
                balance -= event['amount']
    return balance

# Write side: produce commands ke Kafka
producer.produce('commands', value=json.dumps({
    'type': 'create_order',
    'data': {'product': 'laptop', 'qty': 1},
}))
 
# Read side: consume events dan update read model
consumer = KafkaConsumer('events', ...)
for msg in consumer:
    event = json.loads(msg.value)
    if event['type'] == 'order_created':
        db.execute('INSERT INTO order_view ...', event['data'])

# Coordinator publish command
producer.produce('saga', key='saga-123', value=json.dumps({
    'step': 1,
    'action': 'reserve_inventory',
    'data': {'item': 'laptop', 'qty': 1},
}))
 
# Service consume, process, publish next step atau compensating action
# Jika reserve_inventory gagal: publish 'release_payment' (compensating)

# Describe all topics
kafka-topics.sh --bootstrap-server localhost:9092 --describe
 
# Under-replicated partitions
kafka-topics.sh --bootstrap-server localhost:9092 --describe |
  grep "UnderReplicated"
 
# Consumer group lag
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --all-groups |
  awk '{print $1, $2, $5, $6}' | column -t