Apache Kafka Cheatsheet Cheatsheet

🏗️

Core Concepts & Architecture

FUNDAMENTAL

kafka/concepts.sh

# ── Kafka Core Architecture ──
# Producer  -->  Broker (Kafka Cluster)  -->  Consumer
# Producer writes to Topics
# Topics are split into Partitions
# Partitions are replicated across Brokers
# Consumers read from Partitions in Consumer Groups

# Key terms:
# Topic        - Logical category/feed of messages
# Partition    - Ordered, immutable sequence within a topic
# Offset       - Unique identifier per message within a partition
# Broker       - A Kafka server (node in the cluster)
# Cluster      - Multiple brokers working together
# Producer     - App that publishes messages to topics
# Consumer     - App that subscribes to and reads messages
# Consumer Grp - Group of consumers that share load
# ZooKeeper    - Metadata management (KRaft replaces this)
# KRaft        - Kafka Raft mode (ZooKeeper-free, Kafka 3.3+)

Core Components

Component	Role
Broker	Kafka server, stores & serves data
Topic	Message category / feed
Partition	Ordered log segment within a topic
Producer	Writes messages to topics
Consumer	Reads messages from topics
Consumer Group	Set of consumers sharing load
Controller	Broker managing partition leaders
ZooKeeper / KRaft	Cluster metadata & coordination

Key Properties

Property	Description
Durable	Messages persisted to disk
Scalable	Horizontal scaling via partitions
Ordered	Order guaranteed per partition
Fault-tolerant	Partition replication
Fast	O(1) reads/writes with sequential I/O
Distributed	Cluster of multiple brokers

💡

Kafka guarantees ordering only within a partition. If you need global ordering, use a single partition (limits throughput). For most use cases, partition by a key (e.g., order_id) so all events for the same entity go to the same partition.

📁

Topics & Partitions

MANAGEMENT

kafka/topics.sh

# ── Topic Management ──
# Create topic
bin/kafka-topics.sh --create \
  --bootstrap-server localhost:9092 \
  --topic orders \
  --partitions 6 \
  --replication-factor 3

# List topics
bin/kafka-topics.sh --list --bootstrap-server localhost:9092

# Describe topic
bin/kafka-topics.sh --describe \
  --bootstrap-server localhost:9092 \
  --topic orders

# Alter topic (increase partitions — CANNOT decrease!)
bin/kafka-topics.sh --alter \
  --bootstrap-server localhost:9092 \
  --topic orders \
  --partitions 12

# Delete topic
bin/kafka-topics.sh --delete \
  --bootstrap-server localhost:9092 \
  --topic orders

# ── Topic Configuration ──
# Create with configs
bin/kafka-topics.sh --create \
  --bootstrap-server localhost:9092 \
  --topic events \
  --partitions 3 \
  --replication-factor 2 \
  --config retention.ms=86400000 \
  --config cleanup.policy=delete \
  --config compression.type=lz4 \
  --config max.message.bytes=10485760

# Describe configs
bin/kafka-configs.sh --describe \
  --bootstrap-server localhost:9092 \
  --topic events

# ── Partition Reassignment (rebalance) ──
# Generate reassignment plan
bin/kafka-reassign-partitions.sh --generate \
  --topics-to-move-json-file topics.json \
  --broker-list "0,1,2,3"

# Execute reassignment
bin/kafka-reassign-partitions.sh --execute \
  --reassignment-json-file plan.json \
  --bootstrap-server localhost:9092

Partition Strategy

Strategy	Description
Round Robin	Messages distributed evenly (no key)
Key Hashing	Same key always goes to same partition
Custom	User-defined partitioner
Sticky	Batch by key, then round-robin batches

Topic Configs

Config	Default	Description
num.partitions	1	Partition count
default.replication.factor	3	Replica count
retention.ms	604800000 (7d)	Data retention time
retention.bytes	-1 (unlimited)	Max size per partition
cleanup.policy	delete	delete or compact
compression.type	producer	none/gzip/snappy/lz4/zstd
max.message.bytes	1048588 (1MB)	Max message size
min.insync.replicas	1	Min sync replicas for acks=all

⚠️

You can never decrease partitions. Plan partition count carefully. A common formula: partitions = max(throughput_MB/s / partition_throughput, consumer_count). Too many partitions increase latency and memory usage.

📤

Producers

WRITING

kafka/producer.java

// ── Java Producer ──
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

// Reliability configs
props.put("acks", "all");                    // Wait for all ISRs
props.put("retries", Integer.MAX_VALUE);      // Infinite retries
props.put("max.in.flight.requests.per.connection", 5);  // Idempotent order
props.put("enable.idempotence", "true");     // Exactly-once to one partition
props.put("linger.ms", 10);                  // Batch up to 10ms
props.put("batch.size", 16384);              // 16KB batch size
props.put("compression.type", "lz4");        // Compress batches
props.put("delivery.timeout.ms", 120000);    // 2 min total timeout

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

// Fire-and-forget
producer.send(new ProducerRecord<>("orders", "key1", "order data"));

// Synchronous send (wait for response)
try {
    RecordMetadata meta = producer.send(
        new ProducerRecord<>("orders", "key1", "order data")
    ).get();
    System.out.println("Offset: " + meta.offset());
} catch (Exception e) {
    e.printStackTrace();
}

// Asynchronous send with callback
producer.send(new ProducerRecord<>("orders", "key1", "order data"),
    (metadata, exception) -> {
        if (exception != null) {
            exception.printStackTrace();
        } else {
            System.out.println("Sent to partition " + metadata.partition()
                + " offset " + metadata.offset());
        }
    });

// Send with headers
ProducerRecord<String, String> record = new ProducerRecord<>(
    "orders", null, "value"
);
record.headers().add("trace-id", "abc123".getBytes());
record.headers().add("version", "v2".getBytes());
producer.send(record);

// Flush & close
producer.flush();
producer.close();

Producer ACKs

acks	Behavior	Durability
0	No acknowledgment	No guarantee
1	Leader only	Leader crash = data loss
all / -1	All ISRs	Strongest (with min.insync.replicas)

Idempotent Producer

Config	Purpose
enable.idempotence=true	Dedup within partition
acks=all (required)	All ISRs must confirm
max.in.flight=<=5	Maintains ordering with retries
retries=MAX_VALUE	Never give up on transient errors

💡

Always use idempotent producer in production. Set enable.idempotence=true, acks=all, and min.insync.replicas=2. This prevents duplicate messages during retries without killing throughput.

📥

Consumers & Consumer Groups

READING

kafka/consumer.java

// ── Java Consumer ──
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "order-processor");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

// Reliability configs
props.put("auto.offset.reset", "earliest");    // Start from beginning
props.put("enable.auto.commit", "false");      // Manual commit
props.put("max.poll.records", "500");           // Batch size
props.put("session.timeout.ms", "45000");       // Heartbeat interval
props.put("heartbeat.interval.ms", "3000");
props.put("max.poll.interval.ms", "300000");    // 5 min processing time

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

// Subscribe to topic(s)
consumer.subscribe(Arrays.asList("orders", "payments"));

// Or subscribe with pattern
consumer.subscribe(Pattern.compile("orders-.*"));

// Or assign partitions manually (no rebalancing)
consumer.assign(Arrays.asList(
    new TopicPartition("orders", 0),
    new TopicPartition("orders", 1)
));

// Poll loop with manual commit
try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            System.out.printf("partition=%d, offset=%d, key=%s, value=%s%n",
                record.partition(), record.offset(), record.key(), record.value());

            // Process record...
            // If processing fails, seek back to retry
        }
        // Manual commit after successful processing
        consumer.commitSync();   // blocking
        // OR
        consumer.commitAsync();  // non-blocking
    }
} finally {
    consumer.close();
}

// Seek to specific offset
consumer.seek(new TopicPartition("orders", 0), 42);

// Seek to beginning / end
consumer.seekToBeginning(Arrays.asList(new TopicPartition("orders", 0)));
consumer.seekToEnd(Arrays.asList(new TopicPartition("orders", 0)));

// Get consumer group info
Map<TopicPartition, OffsetAndMetadata> committed =
    consumer.committed(new TopicPartition("orders", 0));

Consumer Group Rebalancing

Strategy	Behavior
Range	Partitions divided by range across consumers
RoundRobin	Partitions assigned round-robin
Sticky	Minimize partition movement on rebalance
CooperativeSticky	Incremental, no stop-the-world

Offset Reset Policy

Policy	Behavior
earliest	Read from beginning of partition
latest	Read only new messages
none	Throw exception if no offset found

🚫

Always commit offsets AFTER processing. Use enable.auto.commit=false and commit manually. Auto-commit can cause data loss if the consumer crashes after commit but before processing.

📋

Schema Registry & Serialization

DATA

kafka/schema-registry.sh

# ── Schema Registry REST API ──
# Base URL: http://localhost:8081

# Register a schema (Avro)
curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{"schema": "{\"type\":\"record\",\"name\":\"Order\",\"fields\":[{\"name\":\"order_id\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"double\"}]}"}' \
  http://localhost:8081/subjects/orders-value/versions

# Get latest schema
curl http://localhost:8081/subjects/orders-value/versions/latest

# Get all subjects (registered schemas)
curl http://localhost:8081/subjects

# Get all versions of a subject
curl http://localhost:8081/subjects/orders-value/versions

# Check schema compatibility
curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{"schema": "..."}' \
  http://localhost:8081/compatibility/subjects/orders-value/versions/latest

# Update compatibility mode
curl -X PUT -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{"compatibility": "BACKWARD"}' \
  http://localhost:8081/config/orders-value

kafka/avro-producer.java

// ── Avro Producer with Schema Registry ──
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("schema.registry.url", "http://localhost:8081");
props.put("key.serializer", StringSerializer.class.getName());
props.put("value.serializer", KafkaAvroSerializer.class.getName());

KafkaProducer<String, Order> producer = new KafkaProducer<>(props);

// Generate Avro class from schema (avro-maven-plugin)
// Order.java is generated from order.avsc
Order order = Order.newBuilder()
    .setOrderId("ORD-001")
    .setAmount(99.99)
    .setItems(Arrays.asList("item1", "item2"))
    .build();

producer.send(new ProducerRecord<>("orders", order.getOrderId(), order));

// ── Avro Consumer ──
Properties cProps = new Properties();
cProps.put("bootstrap.servers", "localhost:9092");
cProps.put("schema.registry.url", "http://localhost:8081");
cProps.put("group.id", "order-consumer");
cProps.put("key.deserializer", StringDeserializer.class.getName());
cProps.put("value.deserializer", KafkaAvroDeserializer.class.getName());
cProps.put("specific.avro.reader", "true");  // Use generated class

KafkaConsumer<String, Order> consumer = new KafkaConsumer<>(cProps);
consumer.subscribe(Collections.singletonList("orders"));

ConsumerRecords<String, Order> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, Order> record : records) {
    Order order = record.value();
    System.out.println(order.getOrderId() + ": " + order.getAmount());
}

Serialization Formats

Format	Pros	Cons
Avro	Compact, schema evolution, SR	Schema required
Protobuf	Fast, compact, language-agnostic	Schema required
JSON Schema	Human-readable, flexible	Larger payload
JSON (raw)	No schema needed	No evolution support
String	Simplest	No structure, no validation

Compatibility Modes

Mode	Description
BACKWARD	New can read old data
FORWARD	Old can read new data
FULL	Both backward and forward
BACKWARD_TRANSITIVE	New reads all previous versions
FORWARD_TRANSITIVE	All previous read new
NONE	No compatibility check

🌊

Kafka Streams

PROCESSING

kafka/streams.java

// ── Kafka Streams DSL ──
Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "order-processor");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

StreamsBuilder builder = new StreamsBuilder();

// ── Simple transform: filter + map ──
KStream<String, String> orders = builder.stream("orders");
orders
    .filter((key, value) -> value.contains("completed"))
    .mapValues(value -> value.toUpperCase())
    .to("completed-orders");

// ── Aggregate: count orders per customer ──
KTable<String, Long> orderCounts = orders
    .groupBy((key, value) -> extractCustomer(value),
        Grouped.with(Serdes.String(), Serdes.String()))
    .count(Materialized.as("order-counts-store"));

orderCounts.toStream().to("order-counts");

// ── Join: orders with customer data ──
KTable<String, String> customers = builder.table("customers",
    Consumed.with(Serdes.String(), Serdes.String()));

KStream<String, String> enriched = orders
    .join(customers,
        (order, customer) -> order + " | " + customer,
        Joined.with(Serdes.String(), Serdes.String(), Serdes.String()));

// ── Windowed aggregation: 5-minute tumbling window ──
orders
    .groupByKey()
    .windowedBy(TimeWindows.of(Duration.ofMinutes(5)).grace(Duration.ofMinutes(1)))
    .count()
    .toStream()
    .map((windowedKey, count) -> new KeyValue<>(
        windowedKey.key() + "-" + windowedKey.window().startTime(),
        count.toString()))
    .to("order-counts-5min");

// ── Build and start ──
KafkaStreams streams = new KafkaStreams(builder.build(), props);
streams.start();

// Graceful shutdown
Runtime.getRuntime().addShutdownHook(new Thread(streams::close));

💡

Kafka Streams is a client library — no separate cluster needed. It uses Kafka for input/output and handles failover via consumer group coordination. State stores are changelogged to Kafka topics for fault tolerance.

🖥️

CLI & Administration

OPS

kafka/cli.sh

# ── Console Producer ──
bin/kafka-console-producer.sh \
  --bootstrap-server localhost:9092 \
  --topic orders \
  --property parse.key=true \
  --property key.separator=:

# ── Console Consumer ──
bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic orders \
  --from-beginning \
  --property print.key=true \
  --property key.separator=": "

# ── Consumer Groups ──
bin/kafka-consumer-groups.sh --list \
  --bootstrap-server localhost:9092

bin/kafka-consumer-groups.sh --describe \
  --bootstrap-server localhost:9092 \
  --group order-processor

# Reset consumer offset
bin/kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 \
  --group order-processor \
  --topic orders \
  --reset-offsets --to-earliest \
  --execute

# ── Message inspection ──
bin/kafka-run-class.sh kafka.tools.GetOffsetShell \
  --broker-list localhost:9092 \
  --topic orders

# ── Cluster Info ──
bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092
bin/kafka-metadata.sh --snapshot --command-config client.properties

# ── Log dirs (disk usage) ──
bin/kafka-log-dirs.sh --bootstrap-server localhost:9092 \
  --describe --topic-list orders

Common Admin Tasks

Task	Command
Create topic	kafka-topics.sh --create
Add partitions	kafka-topics.sh --alter
Delete topic	kafka-topics.sh --delete
Describe topic	kafka-topics.sh --describe
List consumer groups	kafka-consumer-groups.sh --list
Reset offsets	kafka-consumer-groups.sh --reset-offsets
Check lag	kafka-consumer-groups.sh --describe
Reassign partitions	kafka-reassign-partitions.sh

Important Metrics

Metric	What to Watch
BytesIn/BytesOut	Throughput (bytes/sec)
MessagesIn/sec	Message throughput
UnderReplicatedPartitions	Should be 0
OfflinePartitions	Should be 0
ConsumerLag	Messages behind consumer
RequestLatency	P99 < 50ms typical
LogFlushLatency	Disk flush time
NetworkProcessorAvgIdle	< 0.3 = overloaded

🎯

Interview Q&A

PREP

Q: How does Kafka guarantee message ordering?Kafka guarantees ordering only within a single partition. Messages with the same key always go to the same partition (via hash(key) % num_partitions). For cross-partition ordering, use a single partition (limits throughput) or buffer and reorder in the consumer.

Q: What is the difference between idempotent producer and transactions?Idempotent producer prevents duplicates within a single partition (PID + sequence number). Transactions provide exactly-once across multiple partitions (atomic writes to multiple partitions). Enable both with enable.idempotence=true and transactional.id.

Q: How does consumer rebalancing work?When a consumer joins or leaves a group, the coordinator triggers rebalancing. Each consumer gets a subset of partitions. With CooperativeSticky, rebalancing is incremental — only affected partitions move. With eager (default), all consumers stop and restart.

Q: What is consumer lag and how do you handle it?Consumer lag = difference between latest offset and consumer offset. High lag means the consumer cannot keep up. Solutions: (1) add more consumers (up to partition count), (2) optimize consumer processing, (3) batch processing, (4) use compacted topics to reduce data volume.

Q: When should you use Kafka vs RabbitMQ?Kafka: high throughput, persistent log, replay, event sourcing, stream processing. RabbitMQ: message queuing, request/reply, complex routing, lower latency, per-message acknowledgment. Kafka is a log, RabbitMQ is a queue.

Q: What is log compaction?Log compaction retains only the latest value for each key, removing older records with the same key. Enable with cleanup.policy=compact. Useful for changelog-style topics where you only care about the current state (e.g., user profile updates).

Q: Explain the role of ZooKeeper / KRaft.ZooKeeper stores cluster metadata (brokers, topics, partition assignments, ACLs). KRaft (Kafka Raft) replaces ZooKeeper entirely, removing the operational overhead of maintaining a separate ZooKeeper ensemble. KRaft is production-ready since Kafka 3.3+.

💡

Top Kafka interview topics: partitioning and ordering, producer acks and idempotence, consumer groups and rebalancing, exactly-once semantics, Schema Registry, log compaction, consumer lag handling, and Kafka Streams vs Kafka Connect.

⏳

Loading cheatsheet...