How this handbook teaches

The goal is not to memorize configuration names. It is to understand what Kafka does at each step well enough to predict its behavior, choose settings on purpose, design with it, and debug it at 2 AM. So the handbook asks you to think before it explains, and to run things before you trust them.

• Try before you read. Every lesson opens with a short question. Answer it on paper first; getting it wrong is what makes the explanation stick.
• Analogy, then mechanism, then trade-off. Each idea starts with the problem it solves and a plain analogy, then how it works, then what it costs. Rarely needed details sit in folded "Go deeper" sections.
• Step through every flow. Diagrams with a step bar reveal one arrow at a time. Predict the next arrow before pressing Next.
• Run it. Labs use the official Apache Kafka Docker image and the command-line tools, so you see offsets, lag, rebalances, and failovers happen. Each lab ends with "Check what you saw".
• Live tools. A key-to-partition explorer that uses Kafka's real murmur2 hash, a partition count calculator, a durability explorer for acks and min ISR, a consumer group playground, a cluster sizing calculator, and a lag drain calculator.
• Notes for work and interviews. Green "In production" and "At work" boxes show how ideas appear in real systems, including payments. Purple "Interview signal" boxes say what interviewers listen for at SDE-2 and what a senior answer adds.

How to study one module

1. Clear the revision queue above at the start of each session: redesign those systems from a blank page.
2. Do the warm-up quiz without reading anything. Note what you missed.
3. For each lesson: answer "Try it first", read, step through the figure, then say out loud what problem the idea solves and when it fails.
4. Do the lab. Predict the output of each command before running it.
5. Do the self-check quiz, then tick lessons as studied.
6. Tick the mastery checklist only when each item is true with your notes closed.

How to do the labs

• Set up once on the lab setup page: a single broker for most labs, a three-broker cluster for replication and failure labs.
• Labs take 15 to 30 minutes. They are written for Kafka 4.3 in Docker, with commands you can paste.
• Tick Done on a lab when you have run it and can explain the "Check what you saw" answers. The dashboard counts finished labs.
• Break things on purpose: stop brokers, kill consumers, reset offsets. That is what the local cluster is for.

How to practice a case study

1. Set a timer for the time on the case study and design it on paper first: requirements, estimates, topics and keys, the diagram, then the hard parts.
2. Open the case study and compare section by section. Answer each guided deep dive before revealing it.
3. Set an honest status at the top of the page. The handbook schedules a review.
4. When it comes due, redesign it from memory, faster each time. Then try related practice prompts in the tracker.

Design statuses and the revision queue

When a design comes up for review, redesign it without notes, then set the new status or press "Reviewed, same status" to restart its timer.

Where your progress is saved

Lesson ticks, lab ticks, mastery checks, readiness checks, and design statuses save automatically to your account, privately: nobody else who opens this page sees them, and they follow you to other devices. The indicator in the top bar shows where the latest change went. If the page cannot reach your account (for example in a private window), it saves in this browser instead, and the tracker has a backup and restore box.

Version and scope

Covered: everything a backend engineer needs to build, run, and debug Kafka-based systems, and what SDE-2 interviews test about Kafka and event-driven design: the data model, producers, consumers, delivery semantics, internals that explain behavior, transactions, schemas, Connect, Streams and Flink at design level, Spring for Apache Kafka, tuning, operations, and troubleshooting.

Left out on purpose: ZooKeeper-mode administration, vendor-specific products in depth, Kafka's source code internals beyond what explains behavior, and non-Java client details (except where they behave differently, such as librdkafka's default partitioner).

The map

Priority overview

Total: about 42 hours including labs, plus about 12 hours for the ten case studies (40 to 50 minutes to design each, then about 30 minutes to compare and note what you missed).

Short paths for specific goals

How each module is organized

• Warm-up quiz to expose what you do not know yet.
• Lessons, each opening with a "Try it first" question, then the idea, an analogy where it helps, how it works, and the trade-offs.
• Figures that step through flows one arrow at a time.
• Labs you run on a local cluster, each ending with "Check what you saw".
• Go deeper folds for details you need only sometimes; In production, At work, and Interview signal notes; Version notes where behavior changed in recent releases.
• Common mistakes, a self-check quiz, and a mastery checklist.

Case studies follow one format: try it first, requirements, estimates, topics and keys, architecture figure, guided deep dives, failures, what a strong answer shows, and follow-up questions.

What you need

• Docker 20.10.4 or newer (Docker Desktop on macOS or Windows, Docker Engine on Linux), with about 4 GB of memory for the three-broker cluster.
• For the Java examples: JDK 17 or newer and Maven or Gradle. Kafka 4.x clients need Java 11 or newer; brokers and tools need Java 17 or newer.
• About 2 GB of free disk.

A single broker

One container runs a combined broker and controller with default settings, listening on localhost:9092.

docker run -d --name broker -p 9092:9092 apache/kafka:4.3.0

Kafka's command-line tools live in /opt/kafka/bin inside the container. Open a shell there and run tools with ./:

docker exec -it --workdir /opt/kafka/bin broker sh
./kafka-topics.sh --bootstrap-server localhost:9092 --list

Stop and delete it when you are done (this also deletes its data):

docker rm -f broker

A three-broker cluster

Used in Modules 3, 5, 12, and 20. Save this as docker-compose.yml. It is the official Apache Kafka combined-mode example with three nodes; each node is both a broker and a controller. Inside the Docker network, nodes talk on port 19092; from your laptop, use localhost:29092, localhost:39092, and localhost:49092.

x-kafka-env: &kafka-env
  KAFKA_PROCESS_ROLES: 'broker,controller'
  KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT'
  KAFKA_CONTROLLER_QUORUM_VOTERS: '1@kafka-1:9093,2@kafka-2:9093,3@kafka-3:9093'
  KAFKA_LISTENERS: 'PLAINTEXT://:19092,CONTROLLER://:9093,PLAINTEXT_HOST://:9092'
  KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT'
  KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
  CLUSTER_ID: '4L6g3nShT-eMCtK--X86sw'
  KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
  KAFKA_LOG_DIRS: '/tmp/kraft-combined-logs'

services:
  kafka-1:
    image: apache/kafka:4.3.0
    hostname: kafka-1
    container_name: kafka-1
    ports: ["29092:9092"]
    environment:
      <<: *kafka-env
      KAFKA_NODE_ID: 1
      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://kafka-1:19092,PLAINTEXT_HOST://localhost:29092'
  kafka-2:
    image: apache/kafka:4.3.0
    hostname: kafka-2
    container_name: kafka-2
    ports: ["39092:9092"]
    environment:
      <<: *kafka-env
      KAFKA_NODE_ID: 2
      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://kafka-2:19092,PLAINTEXT_HOST://localhost:39092'
  kafka-3:
    image: apache/kafka:4.3.0
    hostname: kafka-3
    container_name: kafka-3
    ports: ["49092:9092"]
    environment:
      <<: *kafka-env
      KAFKA_NODE_ID: 3
      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://kafka-3:19092,PLAINTEXT_HOST://localhost:49092'

Start it, then define a helper that runs any tool on kafka-1 against the cluster:

docker compose up -d
k() { tool=$1; shift; docker exec -it kafka-1 /opt/kafka/bin/$tool --bootstrap-server kafka-1:19092 "$@"; }
k kafka-topics.sh --list

The helper puts --bootstrap-server right after the tool name, which every Kafka tool accepts. Internal topics such as __consumer_offsets use replication factor 3 by default, which this cluster satisfies. Stop and remove everything with docker compose down.

A Java project

Add the Kafka client library. The client version can be newer than your brokers; Kafka 4.x clients work with brokers 2.1 and newer.

<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
  <version>4.3.0</version>
</dependency>

implementation 'org.apache.kafka:kafka-clients:4.3.0'

For Spring Boot (Module 18), add org.springframework.kafka:spring-kafka and let Spring Boot manage its version.

Optional tools

• A web UI helps you browse topics and consumer groups. Open-source options include Kafbat UI, AKHQ, and Redpanda Console. None is required; every lab uses the bundled command-line tools.
• kcat (formerly kafkacat) is a fast command-line producer and consumer, handy for scripts.

Command-line tools you will use

Commands vs events. A command asks for something to happen ("SendEmail") and has one intended handler. An event states a fact ("PaymentCompleted") that any number of services can react to. Kafka works best with events, because the producer does not need to know who listens, so adding a consumer never touches the producer.

Kafka is a distributed commit log. It takes this simple structure and splits each log into partitions spread across many servers, with copies on several servers so a crash loses nothing. Databases already use a log internally (the write-ahead log); Kafka makes the log itself the product that other systems read.

• Events are facts: PaymentAuthorized, CardBlocked. They are immutable.
• State is derived: a balance is the sum of its events. Module 13 shows topics that keep only the latest value per key, which turns a log into a table.

Four properties come out of this design:

• Durable: records are written to disk on several brokers before they count as committed.
• Scalable: more partitions spread load across more brokers and more consumers.
• Ordered where it matters: strict order within a partition, chosen by the record key.
• Replayable: records stay for the retention period, so any consumer can re-read history.

• Event-driven microservices: services publish facts; others react. New features subscribe without changing the producer.
• Activity tracking and analytics: clicks, page views, and app events at very high volume.
• Log and metrics pipelines: agents ship logs to Kafka, which buffers them for search and storage systems.
• Data integration and change data capture (CDC): stream every database change to caches, search indexes, and the data warehouse.
• Stream processing: compute running totals, detect fraud patterns in time windows, and join streams as events arrive.
• Audit trail and event sourcing: an ordered, immutable history of what happened.
• Buffering: absorb bursts between fast producers and slow sinks.

• Request-reply that needs an answer now. You can build it on Kafka, but it adds latency and complexity for no gain.
• Per-message delays, priorities, or complex routing rules. RabbitMQ and SQS have these built in; on Kafka you build them from topics (Module 10 and Case study 8).
• Large payloads. The default limit is about 1 MB per record batch. Put big objects in object storage.
• Ad hoc queries. Kafka has no index by key or field. You cannot ask "all events for account 42" efficiently; sink the data into a database for that.
• Global ordering at high throughput. Total order means one partition, which caps throughput at what one partition can handle.
• Tiny workloads or no team to run it. Use a managed service (Confluent Cloud, Amazon MSK, Aiven) or a simpler queue.

• Key and value are bytes. Producers turn objects into bytes with serializers and consumers turn them back with deserializers (Module 15). The key is optional.
• Headers are key-value pairs of metadata, such as a trace ID or an event type.
• Timestamp: by default the producer sets it when it creates the record (CreateTime). A topic can switch to broker time with message.timestamp.type=LogAppendTime. Since Kafka 4.0 the broker rejects CreateTime timestamps more than one hour in the future by default (message.timestamp.after.max.ms).
• Size: the default broker limit is about 1 MB per record batch (message.max.bytes). Large payloads belong in object storage (Module 15).

• A topic is a named stream: payments, card.authorizations.v1. Many producers can write to it and many consumers can read it.
• Topic configs override broker defaults per topic: retention.ms, cleanup.policy, min.insync.replicas, max.message.bytes, and more.
• Create topics explicitly with chosen partitions, replication, and retention. Brokers auto-create topics on first use by default (auto.create.topics.enable=true); most production clusters turn that off so a typo does not create a topic with one partition and one replica.
• Name topics consistently, for example <domain>.<entity>.<event>.v<version>. Names are case-sensitive and cannot be changed later; you create a new topic and migrate.

• Writes only append to the end of a partition. The leader assigns the next offset.
• Each partition replica lives on one broker, so one partition's throughput and size are bounded by one machine. More partitions spread a topic across machines.
• Partition count is chosen at creation. You can increase it later but never decrease it, and increasing it remaps keys (Module 6).
• Consumers: within one consumer group, each partition is read by at most one member at a time, so the partition count caps a group's parallelism (Module 7).

• The producer picks the partition from the key: by default murmur2(key bytes) modulo the partition count (Module 6). Same key, same partition, same order.
• No key means the producer spreads records across partitions with no ordering promise.
• Choose the key as the entity whose events must stay in order: account, card, order, or chat conversation. That gives per-entity order and lets different entities process in parallel.

• An offset is a record's position within its partition. The leader assigns it on append; it only increases and is never reused.
• Offsets can have gaps. Compaction removes records, and transaction markers take up offsets (Module 14). Do not count records by subtracting offsets when exactness matters.
• Consumers own their position. A consumer group commits offsets to Kafka (the internal __consumer_offsets topic) so a restart resumes where it left off (Module 8).
• Consumers can move: to the beginning, the end, a specific offset, or the first offset at or after a timestamp.

• With the default cleanup.policy=delete, records are removed when they are older than retention.ms (7 days by default) or when a partition exceeds retention.bytes (unlimited by default).
• Reading never deletes. Deletion removes whole log segments, so data can stay a little longer than the retention setting (Module 11).
• Retention can be infinite (retention.ms=-1) for event-sourced or audit topics; budget disk or use tiered storage (Module 11).
• cleanup.policy=compact keeps the latest record per key instead (Module 13).

Choose retention from three questions: how long can a consumer be down and still recover? How far back will you replay? What do cost and compliance allow?

docker exec -it --workdir /opt/kafka/bin broker sh

./kafka-topics.sh --bootstrap-server localhost:9092 --create --topic payments \
  --partitions 3 --replication-factor 1
./kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic payments

./kafka-console-producer.sh --bootstrap-server localhost:9092 --topic payments \
  --reader-property parse.key=true --reader-property key.separator=:

acct-5:deposit 100
acct-4:deposit 50
acct-5:withdraw 30
acct-1:deposit 70
acct-5:withdraw 10
acct-4:withdraw 20

./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic payments --from-beginning \
  --formatter-property print.partition=true \
  --formatter-property print.offset=true \
  --formatter-property print.key=true

./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic payments \
  --partition 0 --offset 1 \
  --formatter-property print.offset=true --formatter-property print.key=true

• Each broker has a unique node.id. A broker can host thousands of partition replicas.
• Partitions and their leaders are spread across brokers so load is shared.
• Leaders do the work: producers write to the leader; consumers read from the leader by default. Followers fetch from the leader to stay in sync.

• Replication factor (RF): how many copies of each partition exist, counting the leader. RF 3 is the production norm.
• In-sync replicas (ISR): the replicas that are fully caught up with the leader. A follower that falls behind for longer than replica.lag.time.max.ms (30 seconds by default) is removed from the ISR.
• Failover: if a leader's broker dies, the controller picks a new leader from the ISR, so no committed record is lost.
• The production recipe is RF 3, min.insync.replicas=2, and producers with acks=all. Module 5 explains each part; Module 12 covers replication internals.

• Roles: each node runs as a broker, a controller, or both (process.roles=broker,controller, called combined mode, fine for development but not recommended for critical production).
• Quorum size: run 3 or 5 controllers. A majority must be alive: 3 tolerate one failure, 5 tolerate two.
• Active controller: the Raft leader among the controllers. It writes every metadata change to the __cluster_metadata log; standbys replicate it and can take over immediately.
• Brokers fetch the metadata log to keep their cache current and send heartbeats. A broker that misses heartbeats for broker.session.timeout.ms (9 seconds by default) is fenced, and its partitions get new leaders.

• bootstrap.servers is only for first contact. List two or three brokers, or one DNS name that resolves to several, so a single dead broker does not block startup.
• Clients route themselves. The producer picks the partition and sends to its leader; there is no proxy in the data path. Every client must be able to reach every broker directly.
• Metadata refresh happens periodically (metadata.max.age.ms, 5 minutes) and immediately after errors such as NOT_LEADER_OR_FOLLOWER. Clients retry these errors automatically.

• listeners: where the broker binds, one entry per named listener, for example INTERNAL://:19092,EXTERNAL://:9092,CONTROLLER://:9093.
• advertised.listeners: the address the broker hands to clients in metadata, per listener. It must be resolvable and reachable from the clients that use that listener.
• listener.security.protocol.map maps each name to PLAINTEXT, SSL, SASL_PLAINTEXT, or SASL_SSL. inter.broker.listener.name and controller.listener.names say which listener brokers and controllers use among themselves.

listeners=INTERNAL://:19092,EXTERNAL://:9092,CONTROLLER://:9093
advertised.listeners=INTERNAL://kafka:19092,EXTERNAL://localhost:9092
listener.security.protocol.map=INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT
inter.broker.listener.name=INTERNAL
controller.listener.names=CONTROLLER

Never write to or delete internal topics by hand.

k() { tool=$1; shift; docker exec -it kafka-1 /opt/kafka/bin/$tool --bootstrap-server kafka-1:19092 "$@"; }

k kafka-topics.sh --create --topic orders --partitions 6 --replication-factor 3
k kafka-topics.sh --describe --topic orders

docker stop kafka-2
k kafka-topics.sh --describe --topic orders
k kafka-topics.sh --describe --under-replicated-partitions

docker start kafka-2
k kafka-topics.sh --describe --topic orders
k kafka-leader-election.sh --election-type preferred --all-topic-partitions
k kafka-metadata-quorum.sh describe --status

1. Interceptors (optional) can inspect or modify each record, for example to add tracing headers.
2. Serializers turn the key and value into bytes. A serialization error is thrown from send() itself.
3. The partitioner picks the partition (Module 6). This needs topic metadata; if it is missing, send() waits up to max.block.ms (60 seconds).
4. The record accumulator appends the record to the open batch for that partition. Memory for all batches is capped by buffer.memory (32 MB); when it is full, send() blocks up to max.block.ms and then fails the record with a TimeoutException, reported to the callback and the returned Future.
5. The sender thread drains ready batches, groups them by the broker that leads each partition, and sends one produce request per broker.
6. The response completes each record's Future and runs its callback, on the producer's I/O thread.

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;

public class PaymentProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // acks=all and idempotence are already the defaults; stating them documents intent
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        props.put(ProducerConfig.LINGER_MS_CONFIG, 10);
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");

        try (Producer<String, String> producer = new KafkaProducer<>(props)) {
            ProducerRecord<String, String> record =
                new ProducerRecord<>("payments", "acct-42", "{\"type\":\"DEBIT\",\"amt\":1250}");
            record.headers().add("event-type", "PaymentDebited".getBytes());

            producer.send(record, (metadata, exception) -> {
                if (exception != null) {
                    // retries are exhausted or the error is not retriable
                    System.err.println("send failed: " + exception);
                } else {
                    System.out.printf("stored in %s-%d at offset %d%n",
                        metadata.topic(), metadata.partition(), metadata.offset());
                }
            });
        } // close() sends everything still buffered and waits for responses
    }
}

• Asynchronous with a callback (above) is the normal style. Keep callbacks short: they run on the producer's only I/O thread, so a slow callback delays every other send. Never call flush() inside a callback (Kafka 4.1+ rejects it).
• Synchronous: producer.send(record).get() blocks until the broker answers. It is simple and correct, but it sends one record per round trip. Use it in scripts, tests, or when you must confirm before replying to a caller, and expect much lower throughput.
• One producer per application, shared across threads. Sharing also helps batching.
• flush() sends everything buffered and waits. Use it when you must be sure records are written before a next step, such as committing a consumer offset in a non-transactional pipeline.

• batch.size (16 KB by default) is the maximum size of one partition's batch. A full batch is sent at once.
• linger.ms (5 ms by default since Kafka 4.0, previously 0) is how long a batch may wait for more records before it is sent anyway.
• Under heavy load batches fill before linger.ms expires, so linger adds no latency. Under light load a small linger groups records that would otherwise go one per request.
• Typical high-throughput settings: batch.size 64 KB to 256 KB and linger.ms 10 to 50 ms. Measure; Module 19 covers tuning.

• Compression works on whole batches, so bigger batches compress better: batching and compression reinforce each other.
• The broker keeps the producer's compression when the topic's compression.type is producer (the default). If a topic forces a different codec, the broker must recompress every batch, which costs broker CPU.
• Consumers decompress automatically; nothing to configure.

• delivery.timeout.ms (120 s) is the upper bound on how long a record may take from send() returning to its callback: time waiting in the batch, every retry, and time in flight. When it expires the callback gets a TimeoutException.
• request.timeout.ms (30 s) bounds one request to the broker.
• retries defaults to the maximum integer: in practice delivery.timeout.ms decides when to give up. Retries back off from retry.backoff.ms (100 ms) up to retry.backoff.max.ms (1 s).

./kafka-topics.sh --bootstrap-server localhost:9092 --create --topic perf --partitions 6 --replication-factor 1

./kafka-producer-perf-test.sh --bootstrap-server localhost:9092 --topic perf \
  --num-records 500000 --record-size 1000 --throughput -1 \
  --command-property linger.ms=0 --command-property batch.size=16384

./kafka-producer-perf-test.sh --bootstrap-server localhost:9092 --topic perf \
  --num-records 500000 --record-size 1000 --throughput -1 \
  --command-property linger.ms=20 --command-property batch.size=131072

./kafka-producer-perf-test.sh --bootstrap-server localhost:9092 --topic perf \
  --num-records 500000 --record-size 1000 --throughput -1 \
  --command-property linger.ms=20 --command-property batch.size=131072 \
  --command-property compression.type=lz4

acks=all has been the default since Kafka 3.0. "All" means all replicas currently in the ISR, not all replicas that exist. That is why it needs a floor, which is the next lesson.

• min.insync.replicas is a topic config (with a broker or cluster default) and only matters for producers using acks=all.
• If the ISR is smaller than this number, the leader rejects the write with NotEnoughReplicasException. If the ISR shrinks after the leader appended but before the acknowledgment, the producer gets NotEnoughReplicasAfterAppendException. Both are retriable, so the producer keeps trying until delivery.timeout.ms.
• The standard recipe is replication factor 3 with min.insync.replicas=2: writes continue with one broker down, and every acknowledged record exists on at least two brokers.

• A timeout is ambiguous: the request may have failed, or it may have succeeded with the response lost. The producer must retry to avoid loss, and retries risk duplicates.
• At-least-once is what you get from retries alone: nothing lost, duplicates possible.
• The idempotent producer removes the duplicates that its own retries create.

• On by default since Kafka 3.0. It requires acks=all, retries above 0, and max.in.flight.requests.per.connection of 5 or less. If you set acks to 0 or 1, or retries=0, without explicitly enabling idempotence, the client quietly turns idempotence off (with enable.idempotence=true set explicitly, it refuses to start instead). An in-flight limit above 5 fails at startup unless you explicitly disable idempotence.
• Cost: almost none. Keep it on.
• Scope: one producer instance (one PID), per partition. It does not protect against:
  • your own code catching a failed send and calling send() again (a new record with a new sequence number),
  • a producer that restarts and resends records it was unsure about (a new PID),
  • two producer instances sending the same business event.
• For those cases you need transactions (a stable transactional.id, Module 14) or idempotent consumers keyed by a business ID (Module 10).

• With idempotence on, the broker requires sequence numbers to arrive in order. Batch 2 is rejected until batch 1 is written, so ordering holds with up to 5 requests in flight.
• With idempotence off, the only way to guarantee order with retries is max.in.flight.requests.per.connection=1, which costs throughput. This was common advice before Kafka 3.0.

# durability: wait for every in-sync replica (default since 3.0)
acks=all
# no duplicates from the producer's own retries, ordering kept (default since 3.0)
enable.idempotence=true
# keep retrying transient errors for up to 2 minutes (defaults)
retries=2147483647
delivery.timeout.ms=120000
# up to 5 is safe with idempotence (default)
max.in.flight.requests.per.connection=5

replication.factor=3
min.insync.replicas=2
unclean.leader.election.enable=false

And in your code: handle the callback's exception. A record whose callback reports failure is not in Kafka, whatever the settings say.

k kafka-topics.sh --create --topic ledger --partitions 1 --replication-factor 3 \
  --config min.insync.replicas=3
k kafka-topics.sh --describe --topic ledger

docker stop kafka-3
k kafka-topics.sh --describe --topic ledger

docker exec -it kafka-1 /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server kafka-1:19092 --topic ledger \
  --command-property acks=all --command-property retries=0

docker exec -it kafka-1 /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server kafka-1:19092 --topic ledger --command-property acks=1

• Keyed records: partition = toPositive(murmur2(serialized key bytes)) % numPartitions. Same bytes, same partition.
• Keyless records: the producer sticks to one partition until about batch.size bytes have gone to it, then switches. (Sticky partitioning arrived in Kafka 2.4; the current byte-based, adaptive version in 3.3. Before that, keyless records went round robin.) This builds full batches instead of many tiny ones. With partitioner.adaptive.partitioning.enable=true (the default) it sends more to partitions whose brokers are keeping up.
• The hash runs on bytes, not on your object. Changing the key serializer (for example from String to Avro) changes the partition for the same logical key.
• partitioner.ignore.keys=true makes keyed records use the sticky behavior too. That gives up per-key ordering.

A good key has three properties:

1. It matches the ordering need. Everything that must be processed in order shares the key, and nothing more.
2. It has many distinct values, far more than the partition count, so load spreads evenly.
3. It has no giants. No single value should carry a large share of traffic.

• Symptoms: one partition's consumer lag grows while others are near zero; one broker has much higher bytes in; one consumer instance is at 100% CPU.
• More partitions do not help a single hot key. All of that key's records still map to one partition.
• Fixes, from least to most invasive:
  • Pick a finer key that still meets the ordering need, such as card_id instead of merchant_id.
  • Salt the hot key: merchant-42#0 to merchant-42#7 spreads one key over 8 partitions. You lose per-merchant order, so only do this when downstream can reorder or does not care (such as counting).
  • Route known giants to a dedicated topic with its own scaling.
  • Speed up per-record processing on the hot partition, for example by batching writes downstream.

Type any key to see the partition the Java producer chooses, then simulate many keys and one hot key.

• You can only add partitions, with kafka-topics.sh --alter --partitions N. Existing records never move.
• Keyed topics change mapping: after the change, many keys go to different partitions. Old records of a key stay behind, so during the transition a consumer may handle a key's new records before its old ones.
• Compacted topics are worse: a key's old value sits in one partition and its new value in another, and compaction never reconciles them. Readers that rebuild a table from the topic can end up with a stale value.
• Stream processing joins need both inputs to have the same partition count; changing one breaks the join (Module 17).
• Safe ways to grow a keyed topic:
  • Size it for years of growth up front; partitions are cheap compared with a migration.
  • Create a new topic with more partitions, dual-write or replay into it, and move consumers over once they have caught up.
  • Drain first: stop producers, let consumers finish, add partitions, then resume. Order across the change is then preserved.

Too few partitions cap consumer parallelism and per-topic throughput. Too many cost something too:

• Each partition replica means open files and memory on its broker, and work for the controller during failures and reassignments.
• Producers keep a batch per partition, so many partitions mean smaller, emptier batches for the same traffic, which lowers compression and raises request counts.
• Consumers get more partitions to fetch and more to move in a rebalance.

KRaft handles far more partitions per cluster than the ZooKeeper era did, but the batching cost is unchanged. Most topics are fine with tens of partitions; go into the hundreds only when the throughput math says so.

A custom partitioner is rarely needed. Valid uses: give one huge tenant dedicated partitions, or keep a region's data in partitions whose leaders sit in that region's racks.

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.utils.Utils;
import java.util.Map;

public class TenantPartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        int n = cluster.partitionCountForTopic(topic);
        if ("tenant-big".equals(key)) {
            return 0;                                   // reserved for the giant tenant
        }
        return 1 + Utils.toPositive(Utils.murmur2(keyBytes)) % (n - 1);
    }
    @Override public void configure(Map<String, ?> configs) { }
    @Override public void close() { }
}

Register it with partitioner.class. Every producer writing to the topic must use the same partitioner, or keys scatter.

./kafka-topics.sh --bootstrap-server localhost:9092 --create --topic accounts --partitions 3 --replication-factor 1
./kafka-console-producer.sh --bootstrap-server localhost:9092 --topic accounts \
  --reader-property parse.key=true --reader-property key.separator=:

acct-1:a
acct-2:a
acct-3:a
acct-4:a
acct-5:a
acct-6:a
acct-7:a
acct-8:a
acct-9:a

./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic accounts --from-beginning \
  --formatter-property print.partition=true --formatter-property print.key=true

./kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic accounts --partitions 4

• Consumers send fetch requests to partition leaders. Long polling avoids busy loops: the broker holds a fetch until fetch.min.bytes of data is available (1 byte by default) or fetch.max.wait.ms passes (500 ms).
• The client prefetches in the background, so poll() usually returns records already in memory.

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.List;
import java.util.Properties;

public class BillingConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "billing");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(List.of("payments"));
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
                for (ConsumerRecord<String, String> r : records) {
                    System.out.printf("p%d o%d %s=%s%n", r.partition(), r.offset(), r.key(), r.value());
                }
                consumer.commitSync();   // commit what this poll returned, after processing it
            }
        }
    }
}

• poll(Duration) returns up to max.poll.records (500) records, possibly from several partitions; records from one partition arrive in offset order.
• Process, then commit, then poll again. Module 8 covers commit strategies; Module 10 covers clean shutdown with wakeup().
• Keep the loop moving. If the time between two poll() calls exceeds max.poll.interval.ms (5 minutes), the consumer is considered stuck and leaves the group (Module 9).

• All consumers with the same group.id form one group. Kafka assigns each partition to exactly one member, so members never process the same record at the same time.
• Groups are independent: billing, fraud, and analytics each get every record.
• When members join or leave, partitions move (a rebalance, Module 9).
• Parallelism = min(partitions, consumers) for a classic or KIP-848 consumer group.

• subscribe(Pattern) follows topics that match a regular expression, including new ones as they appear.
• Do not mix subscribe and assign on one consumer.

auto.offset.reset applies only when the group has no committed offset for a partition, or its committed offset no longer exists (deleted by retention):

• Model A, one consumer per thread: run N consumers in the same group (in one process or many). Simple, ordering per partition is automatic, and rebalancing handles failures. Limited by the partition count.
• Model B, a poller plus workers: one thread polls and hands records to a pool. More parallelism than partitions, but you must (1) send each key to the same worker to keep per-key order, (2) commit an offset only when every earlier record of that partition is finished, and (3) pause partitions when worker queues fill. Libraries such as Confluent's Parallel Consumer implement this pattern.
• Share groups (Module 9) are the built-in alternative when per-key order does not matter.

./kafka-topics.sh --bootstrap-server localhost:9092 --create --topic orders --partitions 3 --replication-factor 1
./kafka-producer-perf-test.sh --bootstrap-server localhost:9092 --topic orders \
  --num-records 3000 --record-size 100 --throughput 100

./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders --group billing \
  --formatter-property print.partition=true

./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group billing
./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group billing --members

• Committed offset: stored in Kafka for the group. It is where any member restarts reading this partition, after a crash or a rebalance.
• Position: the next offset this consumer instance will fetch. It runs ahead of the committed offset while records are being processed.
• High watermark: the end of the records that are replicated to all in-sync replicas. Consumers cannot read past it.
• Log end offset: the next offset the leader will assign. Records between the high watermark and the log end are not yet fully replicated.
• Lag = high watermark minus committed offset, per partition. Watch lag as a trend and in time ("the oldest unread record is 4 minutes old"), not only as a count.

• enable.auto.commit=true is the client default. Every auto.commit.interval.ms (5 seconds), the next poll() commits the offsets of records returned by earlier polls. close() commits too.
• Safe enough when you process each batch fully before calling poll() again: a crash replays at most a few seconds of records (at-least-once).
• Unsafe when processing is asynchronous: if you hand records to a thread pool and keep polling, auto-commit may record them as done before the workers finish, and a crash loses them.
• Spring for Apache Kafka turns auto-commit off by default and commits for you after your listener returns (Module 18).

props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

while (running) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
    for (ConsumerRecord<String, String> r : records) {
        process(r);
    }
    consumer.commitAsync((offsets, e) -> {          // fast path, no retries
        if (e != null) log.warn("commit failed, a later commit will cover it", e);
    });
}
// on shutdown or partition revocation: one blocking commit that retries
consumer.commitSync();

• To commit a specific record, commit new OffsetAndMetadata(record.offset() + 1) for its partition.
• CommitFailedException means the group rebalanced and this member no longer owns the partition; its uncommitted records will be reprocessed by the new owner. It usually signals processing that is too slow for max.poll.interval.ms (Module 9).
• Commit frequency trades duplicates against overhead: committing after every record minimizes replays but adds a write per record.

Kafka transactions give exactly-once for read-process-write pipelines that stay inside Kafka (Module 14). When the effect is in another system, make the effect idempotent.

• Idempotent processing: record each event's ID in a table with a unique constraint, in the same database transaction as the effect; a redelivery hits the constraint and is skipped (Module 10).
• Natural idempotency: upserts by key (INSERT ... ON CONFLICT UPDATE), setting absolute values ("balance is 500") instead of deltas ("add 100"), or version checks that ignore older versions.
• Store offsets with the results: write the effect and the next offset in one database transaction, and on startup seek to the stored offset. Use assign() or seek inside a rebalance listener.

• Command line (the group must have no active members):

# preview, then apply: --dry-run is the default, --execute applies
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group billing --topic orders \
  --reset-offsets --to-earliest --dry-run
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group billing --topic orders \
  --reset-offsets --to-datetime 2026-10-17T09:00:00.000 --execute

• In code: seek(partition, offset), seekToBeginning, seekToEnd, and offsetsForTimes to find the offset for a timestamp.
• Expiry: committed offsets of an empty group are deleted after offsets.retention.minutes (7 days by default). A consumer that is off for longer restarts according to auto.offset.reset.

./kafka-producer-perf-test.sh --bootstrap-server localhost:9092 --topic orders \
  --num-records 1000 --record-size 100 --throughput -1
./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group billing

./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group billing --topic orders \
  --reset-offsets --shift-by -10 --dry-run
./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group billing --topic orders \
  --reset-offsets --shift-by -10 --execute
./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group billing --topic orders \
  --reset-offsets --to-earliest --execute
./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group billing

• Membership changes: a consumer starts, shuts down cleanly (it sends a leave request), or crashes.
• Liveness failures: missed heartbeats for session.timeout.ms, or no poll() call within max.poll.interval.ms.
• Metadata changes: partitions added to a subscribed topic, or a new topic that matches a subscribed pattern.

Every rebalance has a cost: partitions pause while they move, and records processed but not yet committed are processed again by the new owner.

• The group coordinator is the broker that leads the group's partition of __consumer_offsets.
• Each rebalance bumps the group's generation. Commits from an older generation are rejected, which is how zombie members are kept from committing.
• The group leader (one of the consumers) runs the assignor locally, which is why all members must agree on the assignor.

The default list is [RangeAssignor, CooperativeStickyAssignor], so groups use Range. To move to cooperative rebalancing, do one rolling restart with only CooperativeStickyAssignor in the list.

• Eager: every member revokes everything, then receives a new assignment. Processing stops group-wide for the whole rebalance.
• Cooperative (incremental): members keep the partitions that are not moving. A first round revokes only what must move; a second round assigns it. Most partitions never pause.
• Cooperative rebalancing makes rolling deploys of large groups dramatically smoother.

• Set a unique, stable group.instance.id per instance (for example the Kubernetes pod name in a StatefulSet).
• A static member that restarts and returns within session.timeout.ms gets its partitions back without a rebalance. Raise the session timeout to cover a normal restart, for example 60 to 120 seconds.
• Trade-off: if the instance truly dies, its partitions wait for the session timeout before moving.

consumer.subscribe(List.of("payments"), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        consumer.commitSync(currentOffsets(partitions));   // last chance to record progress
        flushLocalState(partitions);
    }
    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        loadLocalState(partitions);                         // or seek to offsets stored elsewhere
    }
    @Override
    public void onPartitionsLost(Collection<TopicPartition> partitions) {
        dropLocalState(partitions);                         // already owned by someone else: do not commit
    }
});

• onPartitionsRevoked runs before ownership moves: commit offsets and flush buffered work.
• onPartitionsAssigned runs after: load state, or seek() to offsets you store yourself.
• onPartitionsLost runs when ownership was lost without a chance to hand over (for example after a session timeout); committing then would fail.

• Generally available since Kafka 4.0 and enabled on brokers by default. Clients opt in with group.protocol=consumer; the client default is still classic, and the project plans to make consumer the default in Kafka 5.0.
• The broker computes assignments with server-side assignors (uniform, the default, and range). There is no group leader and no JoinGroup or SyncGroup barrier.
• Fully incremental: each member is told only what to give up and what to take; everyone else keeps working.
• Configuration moves to the broker: with the new protocol, the client settings session.timeout.ms, heartbeat.interval.ms, and partition.assignment.strategy are not used. The broker's group.consumer.session.timeout.ms (45 s) and group.consumer.heartbeat.interval.ms (5 s) apply instead.
• Migration: roll consumers to group.protocol=consumer one at a time; a group converts while online, as long as it uses an assignor without custom metadata. Client-side custom assignors are not supported.

• Production-ready since Kafka 4.2 and enabled by default on new 4.2+ clusters. Use KafkaShareConsumer (or kafka-console-share-consumer.sh) with a group ID.
• Records, not partitions, are handed out. Each fetched record is acquired with a lock (share.record.lock.duration.ms, 30 s by default). The consumer acknowledges it as ACCEPT, RELEASE (try again), REJECT (unprocessable), or RENEW (needs more time).
• Delivery attempts are counted. After share.delivery.count.limit attempts (5 by default) a record is archived instead of redelivered.
• Acknowledgement modes: share.acknowledgement.mode=implicit (default) accepts everything from a poll on the next poll or commit; explicit requires acknowledge() per record.
• Use them for job queues: independent tasks where throughput matters more than order, and where you want more workers than partitions. Keep consumer groups for ordered event streams.

• Keep each poll loop iteration short: lower max.poll.records, or move slow work out of the loop safely.
• Use cooperative rebalancing (CooperativeSticky or the new protocol).
• Use static membership for rolling restarts.
• Shut down cleanly: call wakeup(), commit, close() (Module 10).
• Watch for long GC pauses and CPU throttling in containers, which look like dead consumers.
• Monitor the rebalance rate (rebalance-rate-per-hour on the consumer, consumer-group-rebalance-rate on the coordinator).

./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders --group rebal \
  --command-property group.protocol=consumer