Producers, Consumers, and Brokers

Kafka Has Writers, Readers, and Servers

Every Kafka system has three major roles. Producers write records. Consumers read records. Brokers are Kafka servers that store records, replicate partitions, serve client requests, and coordinate cluster metadata. If you understand how these three roles interact, the rest of Kafka becomes much easier to debug.

A producer does not write to "Kafka" as an abstract cloud. It writes to a specific broker that is currently the leader for a specific partition. A consumer does not receive magical pushes from Kafka. It fetches records from brokers. A broker is not just a network proxy. It stores log segments on disk, handles replication, enforces security, exposes metadata, and participates in cluster leadership.

The reason this module matters more than it looks is that almost every production Kafka incident traces back to a misunderstanding of one of these three roles. "Why did we lose a message?" is almost always a producer acknowledgement question. "Why is processing duplicated?" is almost always a consumer offset-commit-timing question. "Why did clients suddenly time out?" is almost always a broker leadership or metadata question. Learning the mechanics up front means you debug real incidents by reasoning instead of guessing.

A Broker Is the Kafka Server, But It Has Many Jobs

A broker is one Kafka server process. A cluster is a group of brokers. In a production Kafka cluster, each broker usually leads some partitions and follows others. Leadership is per partition, not per topic and not per cluster. Broker 1 might lead orders partition 0, broker 2 might lead orders partition 1, and broker 3 might lead payments partition 0.

Bootstrap servers are only the front door

Kafka clients have a setting called bootstrap.servers. Beginners often think this must list every broker. It does not. It only needs enough reachable brokers for the client to connect and discover cluster metadata. After that, the client learns which brokers lead which partitions and talks to the right brokers directly.

client config:
  bootstrap.servers=broker-1:9092,broker-2:9092

client connects to broker-1
broker-1 returns metadata:
  orders partition 0 leader = broker-3
  orders partition 1 leader = broker-2
  payments partition 0 leader = broker-1

client sends each request to the correct leader

In production, teams list two or three brokers in bootstrap.servers, not all of them, specifically so that a single broker being down at startup does not prevent the client from connecting. Listing every broker is not wrong, it is simply unnecessary — the client only needs one live entry point to bootstrap the rest of its metadata.

The controller broker: who manages the cluster itself

A Kafka cluster needs one broker to make cluster-wide decisions: which broker becomes the new leader when a partition's current leader fails, which brokers are allowed in the in-sync replica set, and how topic and partition metadata changes propagate. That broker is called the controller. There is exactly one active controller per cluster at any time.

Historically, Kafka used an external coordination service called ZooKeeper to elect the controller and store cluster metadata. As of Kafka 3.x and later, clusters run in KRaft mode(Kafka Raft), where a small set of dedicated controller nodes replicate cluster metadata among themselves using the Raft consensus protocol, and Kafka no longer depends on ZooKeeper at all. New clusters you stand up today, including local development clusters, should run KRaft mode — it has fewer moving parts, no separate service to operate, and faster controller failover.

What the controller actually does, mechanically: when a broker hosting a partition leader crashes or is shut down, the controller detects this (brokers send periodic heartbeats to the controller quorum), picks a suitable replacement leader from the partition's in-sync replicas, and pushes that new leadership assignment out to every broker in the cluster. Producers and consumers learn about the change the next time they refresh metadata, which is exactly what Part 09 of this module covers in depth.

Why a dedicated controller role, instead of every broker deciding independently

It might seem simpler for each broker to independently detect a failed peer and just start serving whichever partitions it thinks are now leaderless. This breaks down immediately in any real network: two brokers might both detect the same failure at slightly different times and both decide to become leader for the same partition, a scenario called split-brain. A single, agreed-upon controller avoids this entirely by making leadership decisions in one place and propagating them consistently, so every broker in the cluster converges on the same view of who leads what, rather than each broker guessing independently and possibly disagreeing.

Producers Write Records, But Safe Writing Has Many Steps

A producer is any application that writes records to Kafka. It can be a backend service, data connector, mobile ingestion API, log shipper, IoT gateway, or stream processor. The producer's job is not merely to "send JSON." It must serialize records, choose partitions, batch records, retry temporary failures, and decide what acknowledgement is enough for a write to count as successful.

• ✓Serialize the key and value into bytes.
• ✓Fetch metadata so it knows partition leaders.
• ✓Choose a partition using the key or partitioning strategy.
• ✓Batch records for efficiency.
• ✓Send the batch to the leader broker.
• ✓Retry when safe and configured.
• ✓Report success or failure to the application.

Acknowledgements decide durability risk

The producer's acks setting controls when Kafka confirms a write. Withacks=0, the producer does not wait for confirmation. With acks=1, the leader confirms after writing. With acks=all, the leader waits for the required in-sync replicas. For important business events, acks=all plus appropriate replication settings is the normal direction.

Partitioning: how a producer decides where a record goes

Every record a producer sends must be assigned to exactly one partition of the target topic. If the record has a key, the default partitioner hashes that key and maps it deterministically to a partition — the same key always lands on the same partition, for as long as the topic's partition count does not change. If the record has no key, the producer distributes records across partitions using a sticky, batch-aware strategy: it picks one partition and sends a full batch to it, then picks another partition for the next batch, rather than round-robining record by record, which would defeat the purpose of batching by scattering a handful of records across many partially-filled batches.

# Keyed records: same key -> same partition, every time
send(topic="orders", key="customer-42", value=order1)  -> partition 2
send(topic="orders", key="customer-42", value=order2)  -> partition 2
send(topic="orders", key="customer-91", value=order3)  -> partition 0

# Unkeyed records: sticky batching, not strict round robin
send(topic="orders", key=None, value=event1)  -> batch A, partition 1
send(topic="orders", key=None, value=event2)  -> batch A, partition 1
# batch A fills or lingers out, sent; next batch sticks to a new partition
send(topic="orders", key=None, value=event3)  -> batch B, partition 2

Choosing a key is one of the most consequential decisions a producer makes, because it directly determines ordering guarantees downstream. Keying by customer_id guarantees every event for one customer is processed in order by whichever single consumer owns that partition — exactly the property a per-customer state machine or audit trail needs. Keying by something with very uneven distribution, like a single tenant that accounts for half of all traffic, creates a hot partition that no amount of adding consumers can fix, since a partition is only ever owned by one consumer in a group at a time.

Retries and idempotence at the client level

Producers can retry failed sends automatically, controlled by retries andretry.backoff.ms. Retries are safe against transient errors like a leader election in progress, but naive retries can introduce duplicates or reorder records if multiple requests are in flight at once. Enabling enable.idempotence=true closes this gap: the broker tracks a sequence number per producer session and partition, and silently discards a retried write it has already committed, so retries become safe without any application-level deduplication logic.

Delivery is asynchronous by default

Most Kafka client libraries send records asynchronously: the call to send a record returns almost immediately, and the actual network write happens on a background thread. This is a performance feature, but it is also the single most common source of silently lost data in beginner producer code. If you never inspect the result of a send — a delivery report, a callback, a returned future — a failed write looks identical to a successful one from the application's point of view.

# Silent failure pattern — do not do this
producer.send(topic, key, value)
# no callback, no future check — a broker error here is invisible

# Correct pattern — always attach a delivery callback or check the future
future = producer.send(topic, key, value)
try:
    record_metadata = future.get(timeout=10)
    # confirmed: partition + offset are known and durable per acks setting
except Exception as error:
    log_and_alert("producer delivery failed", error)
    # decide: retry, dead-letter, or fail the request upstream

The distinction matters most at scale: a service sending a few requests per minute might notice a failed send immediately because someone is watching the logs closely. A service sending thousands of records per second with no delivery-result handling can lose a meaningful percentage of its writes during a rough patch of broker instability and never know it happened, because every individual send() call still returned normally — the failure only shows up in the ignored callback or future.

Producers Trade Latency for Throughput on Purpose

A naive producer would send one network request per record. At any real volume this is disastrously inefficient — the fixed cost of a network round trip and a broker append operation dominates the actual work of writing a few hundred bytes. Kafka producers instead group records destined for the same partition into a batch and send the whole batch as one request. Batching is what lets a single producer push millions of records per second without saturating the network with tiny packets.

linger.ms: how long to wait before sending an incomplete batch

A batch fills either when it reaches batch.size bytes or when linger.mselapses since the first record was added to it, whichever comes first. linger.ms is the deliberate trade-off knob: it tells the producer "wait up to this many milliseconds hoping more records show up to fill the batch, even if that means this specific record sits a little longer before being sent." A linger.ms of 0, the default in many clients, sends immediately whenever the network is free — lowest latency per record, but batches end up smaller under light load. A linger.ms of 5-20 lets batches fill up under moderate to heavy load, trading a few milliseconds of added latency for dramatically higher throughput and fewer, larger requests.

# linger.ms = 0 (default in many client libraries)
# record arrives -> sent almost immediately if the network is idle
# under high throughput, natural batching still happens because
# many records arrive faster than one round trip completes

# linger.ms = 10
# record arrives -> producer waits up to 10ms for more records
#   before sending the batch, UNLESS batch.size is reached first
# result under moderate load: fewer, fatter requests to the broker
#   at the cost of up to 10ms of added latency per record

Compression: trading CPU for network and disk

Producers can compress each batch before sending it, using compression.type — common choices are gzip, snappy, lz4, and zstd. Compression happens once, on the whole batch, on the producer side; brokers store the compressed batch as-is and only decompress it when a consumer's client library needs to read individual records back out (the broker itself does not need to decompress to append the batch to the log, which keeps broker CPU usage low). Compressing at the batch level, after records are grouped, gets far better compression ratios than compressing records individually, because similar JSON keys and repeated field names across many records in the same batch compress well together.

The cost of aggressive batching is not just latency. A larger batch.size and higherlinger.ms mean more unsent data sitting in producer memory (bounded bybuffer.memory) at any moment, which means more data at risk if the producer process crashes before that batch is sent. For latency-sensitive paths — an API request that must return a confirmation to a user — teams often keep linger.ms low or zero and accept the smaller batches. For high-volume background ingestion where nobody is waiting on an individual record, raising linger.ms to 10-50ms is a common and safe throughput win.

A worked example: the request-count math

Concrete numbers make the trade-off easier to reason about than settings alone. Consider a producer sending 200,000 small records per second, each around 200 bytes.

Throughput: 200,000 records/sec at ~200 bytes each = ~40 MB/sec

linger.ms = 0, batches average ~20 records (whatever arrives per round trip):
  200,000 / 20 = 10,000 requests/sec sent to the broker

linger.ms = 10, batches average ~2,000 records (filled over 10ms):
  200,000 / 2,000 = 100 requests/sec sent to the broker

Same data volume. 100x fewer requests. Each request now carries
a batch large enough to compress meaningfully well, too --
compounding the throughput gain from Part 04's compression section.

This is why raising linger.ms from 0 to even a small double-digit number is one of the highest-leverage, lowest-risk tuning changes available on a high-volume producer — the added latency per record is small and bounded, while the reduction in broker-side request overhead is often dramatic.

Consumers Pull Records and Own Their Progress

Kafka consumers pull records from brokers. This is different from systems where the broker pushes messages to subscribers. Pull-based consumption lets consumers control their own rate. A fast consumer can fetch more often. A slow consumer can fetch less often. If a consumer is offline, Kafka retains records according to topic retention, and the consumer can catch up later.

A consumer's progress is tracked through offsets. Processing and committing offsets are separate actions. This separation is powerful, but it is also where many reliability bugs begin. If a consumer commits an offset before work is truly complete, a crash can skip work. If a consumer processes work but crashes before committing, the work may happen again after restart.

while running:
  records = consumer.poll()

  for record in records:
    validate(record)
    process_idempotently(record)
    write_result_safely(record)

  commit offsets after successful processing

Offsets themselves are stored durably in Kafka — specifically in an internal, compacted topic called __consumer_offsets. A committed offset is just a message: key is (consumer group, topic, partition), value is the offset and some metadata. This is why offset commits are not free — each commit is itself a write to a Kafka topic, replicated like any other write, which is part of why committing after every single record (rather than after a batch) can become a throughput bottleneck under high volume.

Auto-commit: convenient, and dangerous by default

Most client libraries default to enable.auto.commit=true, which periodically commits the latest offset returned by poll() on a fixed interval (auto.commit.interval.ms, typically 5 seconds), regardless of whether your application has actually finished doing anything useful with those records yet. This is convenient for prototypes and genuinely fine for workloads where occasional reprocessing or occasional loss is acceptable. It is a reliability trap for anything that writes to a database, calls a payment API, or otherwise has a real side effect, because the commit clock runs independently of your processing logic.

enable.auto.commit = true
auto.commit.interval.ms = 5000

t=0s    poll() returns records offset 100-150
t=1s    application starts processing record 100
t=3s    auto-commit fires -- commits offset 150 (already returned by poll)
t=4s    application crashes while still processing record 112
t=restart  consumer resumes from committed offset 150
           records 112-150 are never processed -- silently skipped

The fix is not necessarily to abandon auto-commit everywhere — it is to understand precisely what it commits and when, and to switch to manual, synchronous commits (enable.auto.commit=falseplus an explicit commitSync() or equivalent after your own processing has durably succeeded) for any workflow where a skipped or duplicated record has a real cost.

Sync vs async manual commits

When committing manually, most clients offer both a synchronous and an asynchronous variant.commitSync() blocks until the broker confirms the commit, retrying on retriable errors, and only returns once the offset is durably recorded — the safest choice, at the cost of adding that round-trip latency to your processing loop. commitAsync() fires the commit without blocking and reports success or failure later through a callback, which keeps the loop moving faster but means a failed commit can be missed if the callback is not checked carefully. A common, pragmatic pattern is to use commitAsync() for routine commits during normal processing, and a final commitSync() right before a graceful shutdown, so the last commit is guaranteed to land even if earlier async commits are still in flight.

try:
    while running:
        records = consumer.poll(timeout=1.0)
        for record in records:
            process_idempotently(record)
        if records:
            consumer.commit_async()   # fast, non-blocking, routine path
finally:
    consumer.commit_sync()            # guaranteed final commit on shutdown
    consumer.close()

The Poll Loop Has Its Own Timing Rules

Calling consumer.poll() does more than fetch records. It is also how the consumer tells the broker "I am still alive and working." Kafka consumer groups use this fact to detect dead or stuck consumers and rebalance partitions away from them. Understanding the poll loop's timing settings is what separates a consumer that runs for months without a rebalance storm from one that gets kicked out of its group every few minutes under load.

max.poll.records: how much work one poll hands you

max.poll.records caps how many records a single call to poll() returns. It does not cap how many records exist to be read — it only limits the batch size handed to your application per call. This matters because the consumer must call poll() again withinmax.poll.interval.ms or it is considered dead and removed from the group. If your processing logic is slow per record and max.poll.records is set high, you can end up processing for longer than max.poll.interval.ms allows, triggering a rebalance in the middle of your own processing.

max.poll.records = 500
max.poll.interval.ms = 300000  (5 minutes, the default)
average processing time per record = 700ms

worst case time to process one poll's batch:
  500 records * 700ms = 350,000ms = ~5.8 minutes

5.8 minutes > 5 minute max.poll.interval.ms
-> consumer is presumed dead, group rebalances mid-batch
-> in-flight work may be duplicated by whichever consumer
   picks up the reassigned partition

session.timeout.ms and heartbeats: detecting a dead consumer

Separately from the poll call itself, most Kafka client libraries run a background heartbeat thread that pings the group coordinator broker every heartbeat.interval.ms. If the coordinator does not hear a heartbeat within session.timeout.ms, it assumes the consumer has crashed and triggers a rebalance, even if the consumer's main thread is still technically alive but stuck. This is a separate failure mode from the poll-interval timeout above:session.timeout.ms catches a genuinely frozen or network-partitioned process, while max.poll.interval.ms catches a process that is alive and heartbeating but stuck doing slow application work between polls.

The practical takeaway: if you see unexplained rebalances, check which timeout tripped. A trippedsession.timeout.ms usually points to network issues, GC pauses, or resource starvation on the consumer host. A tripped max.poll.interval.ms almost always points to slow business logic inside the poll loop relative to max.poll.records.

session.timeout.ms = 10000       (10 seconds)
heartbeat.interval.ms = 3000     (3 seconds)

t=0s    consumer sends heartbeat, coordinator resets its timer
t=3s    consumer sends heartbeat, coordinator resets its timer
t=6s    consumer sends heartbeat, coordinator resets its timer
t=8s    a long GC pause freezes the whole JVM, including the
        background heartbeat thread
t=9s    coordinator has not heard a heartbeat since t=6s -- still
        within the 10s window, no action yet
t=17s   GC pause finally ends, but 11 seconds have now passed
        since t=6s -- past the 10s session.timeout.ms
t=17s   coordinator has already declared this consumer dead and
        triggered a rebalance; another consumer now owns its
        partitions

This is exactly why session timeout tuning and JVM garbage collection tuning are often discussed together for Java-based Kafka consumers — a heartbeat thread that shares a process with a long GC pause is just as unresponsive during that pause as a genuinely crashed process, from the coordinator's point of view.

Consumer Groups Let Readers Share Work

A consumer group is a named team of consumers. Kafka assigns partitions to consumers within the group. For any one partition, only one consumer in the same group reads it at a time. This lets a service scale horizontally without every instance duplicating the same work.

orders topic has 4 partitions

billing group:
  billing-consumer-1 -> partition 0, partition 1
  billing-consumer-2 -> partition 2, partition 3

analytics group:
  analytics-consumer-1 -> partition 0, partition 1, partition 2, partition 3

Billing and analytics are separate groups.
They read the same topic independently.

This is why Kafka can behave like a queue and a publish-subscribe system at the same time. Inside one consumer group, partitions are shared, so records are distributed across workers. Across different consumer groups, each group gets its own independent read position, so the same records can feed many applications.

Adding or removing a consumer from a group — including a crash, a deploy, or an autoscale event — triggers a rebalance: the group coordinator reassigns partitions among whichever consumers are currently alive. During a rebalance, consumers involved briefly stop processing while new assignments are handed out. Newer cooperative rebalancing strategies reduce this pause by only reassigning the partitions that actually need to move, instead of revoking every partition from every consumer and reassigning from scratch, which is what the older eager rebalancing protocol does.

Eager vs cooperative rebalancing

Under the original, eager rebalancing protocol, every consumer in the group revokes all of its assigned partitions the moment a rebalance starts, even partitions that will be reassigned right back to the same consumer, and the group only resumes processing once every member has a full new assignment. This is simple to reason about but means the entire group pauses on every membership change, however small. The cooperative (incremental) rebalancing protocol, the default in current client versions, instead computes the minimal set of partitions that actually need to move and only revokes those, letting every other consumer keep processing its unaffected partitions throughout the rebalance.

Static membership: avoiding unnecessary rebalances entirely

A rolling deploy that briefly stops and restarts each consumer instance can trigger a full rebalance for every single restart, even though the group's real membership is unchanged a few seconds later. Setting group.instance.id to a stable, unique value per consumer instance enables static membership: the group coordinator recognizes a consumer reconnecting with the same instance ID within session.timeout.ms as the same member returning, rather than as a departure followed by a new arrival, and skips the rebalance entirely. This is a meaningful operational win for any consumer group that gets redeployed frequently.

Why Kafka Chose Pull Over Push

Many older messaging systems push messages to subscribers: the broker decides when to send data, and the consumer's job is to keep up. Kafka deliberately inverted this. The consumer decides when to ask for more data by calling poll(). This single design choice explains a large share of Kafka's operational behavior, so it is worth understanding in depth rather than as a trivia fact.

The problem push-based systems run into

In a push model, the broker must decide a send rate for each consumer without knowing that consumer's true current capacity. If the broker guesses too high, it floods a slow consumer faster than it can process, forcing the consumer to either drop messages, buffer them until it runs out of memory, or apply some kind of flow-control signal back to the broker asking it to slow down — which is effectively reinventing a pull model through a side channel. If the broker guesses too low, fast consumers sit idle waiting for data that could have already been sent.

With pull, backpressure is implicit and safe by default: a slow consumer simply callspoll() less often or processes what it receives more slowly, and the unread records just sit durably in the broker's log until the consumer is ready for them, bounded only by the topic's retention period. Nothing needs to be buffered in the consumer's memory beyond what it asked for. Nothing needs to be dropped. The broker does not need to track each consumer's real-time capacity — it only needs to serve whatever range of offsets each consumer's next fetch request asks for.

Pull also enables batching on the consumer's own terms

Because the consumer initiates each fetch, it can ask for as many records as it can comfortably handle in one round trip (bounded by max.poll.records and related fetch-size settings covered in Part 06), rather than receiving records one at a time as a broker decides to push them. This is part of why a single Kafka consumer can sustain very high throughput — each network round trip can carry a large batch of records the consumer itself decided it wanted.

What "falling behind safely" looks like in numbers

If a producer sustains 50,000 records/second and a consumer can only sustain 40,000 records/second, a push-based system without its own flow control would either drop the excess 10,000 records/second or force the consumer to buffer them in memory until it runs out. A pull-based consumer instead simply falls 10,000 records/second further behind in the broker's durable log every second — fully visible as growing consumer lag, and fully recoverable later, bounded only by the topic's retention window, exactly as covered in this module's monitoring guidance in Part 10.

Clients Must Know Which Broker Leads Which Partition

Kafka clients are metadata-aware. This is different from a simple load-balanced HTTP service where any backend can handle any request. In Kafka, writes and reads for a partition go through the current partition leader. If leadership changes because a broker fails or maintenance occurs, clients refresh metadata and send future requests to the new leader.

This explains common errors like NotLeaderOrFollower or temporary timeouts during broker restarts. The client may have old metadata for a short period. Good clients refresh and retry. Good applications still log and monitor these errors so real cluster instability is not ignored.

How brokers actually coordinate as a cluster

The controller broker described in Part 02 is what makes leadership changes cluster-wide and consistent rather than something each broker figures out independently. In KRaft mode, cluster metadata — which topics exist, how many partitions each has, who currently leads each partition, which brokers are in each partition's ISR — is itself stored as a replicated, ordered log, managed by the Raft protocol among the controller nodes. Every broker in the cluster maintains a local, continuously updated copy of this metadata log, which is what lets any broker answer a client's metadata request correctly and quickly, without having to ask another broker first.

1. broker-3 (leader for orders partition 0) crashes
2. controller quorum detects the missed heartbeat
3. controller checks orders-partition-0's ISR: [broker-1, broker-2]
4. controller elects broker-1 as the new leader
5. controller writes this change to the cluster metadata log
6. every broker replicates the updated metadata log
7. producer's next write to orders partition 0 is rejected by the
   old cached leader info (NotLeaderOrFollower) or times out
8. producer refreshes metadata, learns broker-1 is now the leader
9. producer resumes sending to broker-1 -- no data was lost because
   acks=all had already required broker-2's replica to be caught up

This is also why acks=all and a correctly sized in-sync replica set matter beyond just the producer-side durability story from Part 03 — they are what guarantee that whichever replica the controller picks as the new leader after a failure is actually caught up, so a leadership change does not silently roll back recently acknowledged writes.

What clients actually cache, and when they refresh it

A client does not fetch fresh metadata before every single request — that would defeat the purpose of knowing partition leaders in the first place. Instead, it caches the metadata it receives and only refreshes it on specific triggers: periodically on a fixed interval (metadata.max.age.ms), immediately when a request to a cached leader fails with an error indicating that broker is no longer the leader, and on startup or when a new topic is referenced that the client has no cached metadata for at all.

This caching-with-triggered-refresh design is why a brief broker restart is usually invisible to a well-behaved client beyond a short burst of retried requests — the client does not need to be told proactively that something changed; it discovers this the moment it tries to use stale information and gets corrected.

Monitoring the Whole Picture: Producers, Brokers, Consumers

Every setting covered in this module maps to a real, observable metric. Treating producers, brokers, and consumers as three independently healthy systems misses the point — a Kafka pipeline is only as reliable as the weakest link across all three, and the right monitoring makes that weak link visible before it becomes an incident rather than after.

Five Misconceptions About Kafka Clients

What This Looks Like on Day One

5 Interview Questions — With Complete Answers

The Mistakes That Make Kafka Clients Unreliable

Errors You Will Hit — And Exactly Why They Happen