Configuring Apache Kafka for High-Load Systems

Introduction

"Millions of messages per second" is a meaningless figure without a load profile: a cluster pushing 2 million 200-byte messages and a cluster pushing 50,000 100 KB messages are constrained by completely different resources. Before touching any parameter, pin down the average and maximum message size, the target volume in MB/s, the required end-to-end p99, and how much loss you can tolerate. Without those four numbers, tuning is guesswork.

The second common cause of failed tuning is carrying over advice from articles written in 2016–2020. Since then the defaults for acks, session.timeout.ms and replica.lag.time.max.ms have changed, and so has the cluster's operating mode entirely. All values below are for Kafka 3.x–4.x, and where a default changed, it is noted in place.

Kafka Streams, Kafka Connect, Schema Registry and multi-region deployments are deliberately out of scope.

Broker settings that affect throughput and latency

Threads and request queues

A Kafka broker keeps two thread pools. num.network.threads (default 3) are the network threads: they read and write sockets and put requests on a queue. num.io.threads (default 8) are the handlers: they write to the log and serve reads.

Network threads do almost no useful work, so scaling them by core count is pointless — on a 64-core machine, three dozen network threads will sit idle. For num.io.threads the guideline is different: the lower bound is the number of disks holding data, the upper bound is the number of cores.

Saturation is visible over JMX: NetworkProcessorAvgIdlePercent (kafka.network:type=SocketServer) and RequestHandlerAvgIdlePercent (kafka.server:type=KafkaRequestHandlerPool), an idle fraction from 0 to 1. Approaching zero means the threads are fully loaded and adding more will help — provided there is spare CPU and the disks aren't the bottleneck. Staying above 0.3 means adding more is pointless and the bottleneck is elsewhere.

Both parameters are dynamic (KIP-226): they can be changed via kafka-configs.sh --alter --entity-type brokers without restarting the broker. An experiment costs seconds, not a maintenance window.

queued.max.requests (default 500) caps the queue between the two pools. When it fills, the network threads stop reading from sockets — the channels are muted, which acts as built-in backpressure for producers and replicas. Raising it smooths out short spikes but increases latency and memory use; watch RequestQueueSize.

Message size limits

socket.request.max.bytes is the maximum size of a single request to the broker, 104857600 by default (100 MiB). Large batches or large individual messages will hit it and be rejected; raise it with an eye on broker memory. message.max.bytes on the broker and max.message.bytes on the topic set the maximum record size, around 1 MiB by default.

A key detail people often forget: the limit applies to the entire compressed record batch, not to an individual message. That is why RecordTooLargeException is usually thrown not because of one oversized message but because of a batch that grew.

To accept larger messages you raise message.max.bytes together with socket.request.max.bytes, replica.fetch.max.bytes (how much a follower replica requests from the leader, 1048576 by default), the producer's max.request.size (1048576), and the consumer's max.partition.fetch.bytes.

There is also a piece of outdated advice still circulating: "if replica.fetch.max.bytes is smaller than message.max.bytes, replication will stall." That was only true before 0.10.1. Since KIP-74 the broker always returns at least one record batch even if it exceeds the fetch limit, so neither replication nor consumption gets stuck on a large message. Keeping the values aligned is still worth doing — for predictability and correct memory math, not out of fear of a deadlock.

Fetch memory is the product of fetch size, number of partitions and number of fetcher threads, so generous limits on a cluster with thousands of partitions eat heap invisibly. And strategically: pushing large objects through Kafka is an anti-pattern. The right shape is claim-check — put a reference to an object in external storage into the topic, not the object itself.

Log segments and flush

log.segment.bytes sets the segment size after which the broker rolls the log; the default is 1 GiB. Segments also roll by time, controlled by log.roll.ms and log.roll.hours (7 days by default), which matters for low-traffic topics where a segment may not reach a gigabyte for weeks.

Large segments reduce file-management overhead but lengthen recovery after an unclean shutdown — on startup the broker verifies the last segment of every partition. num.recovery.threads.per.data.dir speeds that phase up: it defaults to 1, and on multi-disk nodes it makes sense to raise it to the number of disks. Small segments recover faster and are deleted sooner by retention, at the cost of more frequent file open and close.

A note on disk sizing: retention (log.retention.ms, log.retention.bytes) only deletes whole inactive segments. With a 1 GiB segment the actual on-disk footprint is always noticeably larger than what you configured, and headroom has to account for that.

Kafka does not fsync on every write — data is buffered in the page cache and flushed by the OS in the background. log.flush.interval.messages (default Long.MAX_VALUE) and log.flush.interval.ms (unset) let you force flushes, but they are effectively off, and the documentation explicitly recommends leaving them alone and relying on replication instead. Frequent forced fsyncs hurt both latency and throughput noticeably. In production these parameters are left at their defaults.

Network buffers

socket.send.buffer.bytes and socket.receive.buffer.bytes set the broker's TCP buffer sizes, 102400 bytes (100 KiB) by default — enough for a local network. On high-RTT links the buffer is raised based on the bandwidth-delay product; for WAN replication there is a separate replica.socket.receive.buffer.bytes.

A critical detail without which the advice doesn't work: raising the buffer in Kafka's config is useless unless the system limits net.core.wmem_max and net.core.rmem_max are raised too — the kernel will silently truncate the requested size. The value -1 means "use the OS default", and on modern kernels with TCP autotuning that is often a better choice than tuning by hand.

This is also the place to mention zero-copy (sendfile), which underpins Kafka's performance: data moves from the page cache into the socket without passing through user space. With SSL/TLS enabled, zero-copy stops working — every byte goes through the JVM to be encrypted, and CPU load rises noticeably. Budget for that during sizing rather than discovering it after TLS goes live.

Replication threads

num.replica.fetchers sets how many threads a follower broker uses to pull data from each leader, and defaults to 1. On clusters with many partitions this is most often the replication bottleneck: a single thread per source cannot keep up with hundreds of partitions, and the result is non-zero UnderReplicatedPartitions and growing replica lag — while CPU, network and disks all look underutilised.

A reasonable range for busy clusters is 2 to 8 threads, with an eye on core count. The companion settings replica.fetch.min.bytes and replica.fetch.wait.max.ms do for replication what fetch.min.bytes and fetch.max.wait.ms do for the consumer, allowing larger fetches.

Replication has to be tuned together with partition count: growing partitions without growing fetcher threads predictably degrades ISR stability.

Quotas and cluster protection

In a multi-tenant cluster, quotas are the only built-in mechanism that protects brokers from a client that has gone off the rails. They are set through kafka-configs.sh --alter --entity-type clients (or users) and include producer_byte_rate and consumer_byte_rate (bandwidth in bytes per second), request_percentage (the share of broker thread time a client may occupy) and controller_mutation_rate (protection against a flood of topic and partition creation and deletion).

Kafka doesn't drop requests when a quota is exceeded — it delays the response, and that delay shows up in ThrottleTimeMs. That is the first metric to check when a client complains about latency that grew for no apparent reason.

Connection limits belong here too: max.connections, max.connections.per.ip and connections.max.idle.ms (10 minutes by default). With thousands of clients, connection and file-descriptor limits are hit before the thread pools are.

Producer configuration for efficient writes

Acknowledgements (acks)

acks determines how many replicas must acknowledge a write before the leader responds to the producer. acks=0 — don't wait at all; acks=1 — the leader only; acks=all (or -1) — every replica in the ISR.

Since Kafka 3.0 the default is acks=all, not acks=1 as it used to be: KIP-679 turned on producer idempotence by default (enable.idempotence=true), and that requires acks=all. Plenty of articles and notes still list acks=1 as the default — that information is stale.

With acks=1 the leader writes the message to its local log and acknowledges immediately, without waiting for replication. The acknowledgement is faster, but if the leader dies before replicating, the message is gone. Note that lowering acks does not speed up delivery to readers: the point at which a record becomes visible to a consumer is governed by the high watermark advancing across all ISR replicas, and it does not depend on acks at all.

Practical takeaway: don't lower acks "just in case". On modern hardware the throughput difference between acks=all and acks=1 is usually a few percent, not a multiple — measure it on your own load profile first. If you need maximum durability, keep acks=all paired with min.insync.replicas; acks=1, and certainly acks=0, only belong where losing some data is acceptable to the business.

Batching: batch.size and linger.ms

The producer sends messages in batches. batch.size (default 16384, i.e. 16 KiB) sets the target batch size in bytes, linger.ms (default 0) sets the maximum delay before sending in order to accumulate more messages. In Kafka 4.0 the linger.ms default was revisited (KIP-1030) — check the documentation for your version.

A fundamental detail that usually gets missed: batch.size is a per-partition limit, not a per-request one. The producer accumulates a separate batch for every destination partition and packs the ready batches into a single request.

Two consequences follow. The producer's peak memory use is on the order of "number of active partitions × batch.size", and it has to fit inside buffer.memory. And the total request size is capped by max.request.size (1048576 bytes by default), which makes "raise batch.size to 200 KB and leave max.request.size alone" a classic first-tuning mistake.

For high throughput the values go up: batch.size to 100,000–200,000 bytes, linger.ms to 5–50 ms. A bigger batch buys throughput at the cost of added latency for the first messages in the batch.

Verify the result with client metrics rather than by eye: batch-size-avg and record-queue-time-avg. If the average batch is meaningfully smaller than batch.size, the limiting factor is the arrival rate, not the size, and raising batch.size further does nothing.

Compression

compression.type selects the algorithm: gzip, snappy, lz4, zstd or none (default none). lz4 gives the best speed balance; zstd gives a noticeably better ratio at comparable CPU cost, and in recent versions the level is controlled by compression.zstd.level.

A mandatory condition, without which compression turns into an anti-optimisation: on the broker and on the topic, compression.type must be producer (which is the default). If a specific codec is set there and it differs from the one the data arrived with, the broker will decompress and recompress every batch — the most expensive operation you can impose on it, and it reliably eats the entire gain.

The compression ratio depends directly on batch size, because the whole batch is compressed as a unit. So compression.type cannot be tuned in isolation from batch.size and linger.ms: with linger.ms=0 and small batches the gain is minimal. compression-rate-avg is a convenient way to watch the effect.

For very low latency or very small messages the gain may not repay the CPU cost, but in most high-load scenarios producer compression is turned on.

Producer buffer

buffer.memory (default 33554432 bytes, 32 MiB) is the memory available for the send queue. When the buffer fills, send() blocks, and after max.block.ms (60000 ms by default) it throws.

The sizing guideline is concrete: the buffer should be at least "number of active partitions × batch.size", multiplied by 1.5–2 to cover in-flight requests and overhead. With 500 partitions and batch.size=100000 the smallest sensible value is already around 50 MB, so the default falls short by a wide margin.

Whether you hit the right size is shown by buffer-available-bytes (should not regularly drop to zero) and waiting-threads (normally zero). Those metrics, not guesses, tell you whether the application is hitting the buffer ceiling.

In-flight requests

max.in.flight.requests.per.connection (default 5) determines how many requests the producer keeps in flight on a single connection without waiting for acknowledgements. A higher value fills the network better, especially on links with significant RTT.

A persistent misconception needs clearing up here. It is often written that an idempotent or transactional producer is forced to run with max.in.flight=1 — that has been false since Kafka 1.0.0 (KAFKA-5494). An idempotent producer preserves record order at values up to and including 5: the broker tracks sequence numbers and discards or reorders retries itself.

The real constraint is "no more than 5", and it is hard. With idempotence enabled — which it is by default since 3.0 — a value above 5 makes the producer fail to start and throw ConfigException. So the widespread advice to "raise it to 5–10 for throughput" doesn't speed anything up on modern versions; it fails the application at initialisation.

Setting max.in.flight=1 for strict ordering is no longer necessary either — that is a legacy recipe for a non-idempotent producer, and today it only cuts throughput for nothing in return.

And it's worth understanding what idempotence actually buys: no duplicates from retries within a producer session and a single partition. That is not end-to-end exactly-once for the whole pipeline — that needs transactions and matching consumer configuration.

Timeouts and retries

Steady-state throughput is only half the problem; the other half is how the producer behaves when a broker responds slowly or falls over.

The upper bound on a record's life is delivery.timeout.ms (120000 ms by default): the total budget from the send() call to final success or final failure, covering time in the buffer, all retries and network delays. retries defaults to Integer.MAX_VALUE on modern versions and means almost nothing on its own — retries are cut off by delivery.timeout.ms, so that is what you tune, not the attempt count.

request.timeout.ms (30000 ms by default) bounds the wait for a single response, and retry.backoff.ms (100 ms by default) sets the pause between attempts. The relationship has to be sensible: delivery.timeout.ms must be at least the sum of linger.ms and request.timeout.ms, or the configuration is rejected.

An aggressively short delivery.timeout.ms turns a brief degradation of one broker into a flood of application errors; an excessively long one turns it into quiet queue growth and a latency spike. Watch record-error-rate and record-retry-rate to see whether the producer is living on retries.

Partitioning and how full batches get

Actual batch size is determined not only by batch.size and linger.ms but by how records are spread across partitions. For messages without a key, older client versions distributed records round-robin, so with many partitions the batches never filled up and all the batching tuning came to nothing.

Since Kafka 2.4 (KIP-480) there is a sticky partitioner: the producer stays on one partition until it has a full batch and only then switches. This is the change that often yields more throughput than any manual tuning of batch.size. On 3.3 and later, leave partitioner.class unset so the built-in implementation with uniform sticky distribution is used.

For keyed messages the partition is computed from the key's hash, which gives ordering within a key — and that is exactly why any change to a topic's partition count moves keys to different partitions and breaks that guarantee. Partition count is planned up front.

Consumer configuration for fast processing

A consumer keeps parallel fetch requests to every broker holding its partitions, accumulates the responses in a client-side buffer, and hands them to the application in chunks of max.poll.records. What arrives over the network and what is handed to the application are two different boundaries set by different parameters, and confusing them is expensive.

Fetch size and wait

fetch.min.bytes sets the minimum amount of data the broker will gather before answering a fetch request. It defaults to 1 byte — meaning the broker answers immediately even when there is almost nothing to send. Raising it to tens or hundreds of kilobytes forces a proper batch to accumulate and cuts both request frequency and per-message overhead on the client and the broker.

fetch.max.wait.ms bounds the wait from above and defaults to 500 ms: if the volume hasn't accumulated, the broker returns whatever it has. The parameter actually worth changing here is fetch.min.bytes — fetch.max.wait.ms is already 500, so the advice to "raise it to 500" changes nothing.

The price of larger fetches is tail latency: with an uneven stream, delivery p99 rises to roughly fetch.max.wait.ms, so for latency-critical streams it is lowered instead. Where messages arrive densely, raising fetch.min.bytes barely affects latency at all — there is always data to answer with.

Maximum fetch size

max.partition.fetch.bytes limits how much data comes from one partition per request (1048576, 1 MiB by default), and fetch.max.bytes limits the size of the whole broker response (52428800, about 50 MiB). Together they stop one hot partition from taking the entire channel.

As with replication, the advice "make sure to raise max.partition.fetch.bytes to the size of your largest message or consumption will stall" is out of date. After KIP-74 the broker returns at least one record batch even if it exceeds the limit, so a consumer doesn't get stuck on a large message. Raising both parameters is worthwhile for a different reason: fetching more per request and making fewer network round trips.

Memory has to be estimated carefully, though. fetch.max.bytes bounds the response of one broker, while the consumer holds parallel requests to every broker that hosts its partitions. A realistic upper bound on the peak buffer is "number of brokers with assigned partitions × fetch.max.bytes", not "number of partitions × max.partition.fetch.bytes" as is sometimes written: the second formula systematically underestimates consumption. On high-latency links, also look at the consumer's receive.buffer.bytes.

Auto-commit and offsets

enable.auto.commit (default true) and auto.commit.interval.ms (default 5000 ms) decide whether the consumer commits offsets on its own and how often.

The mechanics matter more than the fact. In the classic consumer, auto-commit is not performed by a separate background thread but inside the poll() call: on each poll() the client checks whether the interval has elapsed and commits the offsets from the previous fetch. A non-obvious consequence follows — while your thread is busy with long processing and isn't calling poll(), nothing is committed, however much time passes. It is precisely the picture of auto-commit as an independent background process that produces most of the surprises with losses and duplicates. In the new asynchronous consumer that arrived with the KIP-848 protocol some of this work really has moved to a background thread — one more reason to pin your version.

On failure, auto-commit leaves you uncertain in both directions. If the consumer dies after processing but before committing, the messages count as unread and another instance in the group processes them again. If the commit lands before processing finished and the consumer then dies, some messages are marked read and lost. Critical systems turn auto-commit off (enable.auto.commit=false) and commit explicitly after processing a batch. Committing too often loads the internal __consumer_offsets topic, so intervals of a few seconds are usually right.

With manual control you have commitSync() and commitAsync(): the synchronous one blocks the thread until acknowledged but guarantees offsets are stored; the asynchronous one doesn't delay processing but may fail to confirm the last offsets on a crash. The common pattern combines both — regular commitAsync() inside the processing loop and a final commitSync() before shutdown or in the rebalance handler (ConsumerRebalanceListener.onPartitionsRevoked). Neither on its own gives exactly-once: if processing has side effects, the application must be idempotent or use transactions.

poll and session timeout

A consumer has to call poll() regularly — both to get data and to prove it is alive. If poll() isn't called more often than max.poll.interval.ms (300000 ms, 5 minutes by default), the consumer decides it is stuck and leaves the group itself, sending LeaveGroup and triggering a rebalance. That is the client's doing, not the broker's, so looking for the cause in broker logs is futile.

When processing is slow there are two ways out: raise max.poll.interval.ms, or shrink the chunk with max.poll.records (500 by default). Here two things that are often conflated need separating cleanly. fetch.* controls how much data comes over the network and sits in the client buffer; max.poll.records merely slices the already-fetched buffer into chunks for the application. Lowering max.poll.records shortens one iteration and reduces the risk of overrunning max.poll.interval.ms, but it does not reduce network traffic or memory use — using it to cure an OOM is pointless; fetch.max.bytes and max.partition.fetch.bytes exist for that.

Failure detection is configured separately. session.timeout.ms decides how long without a heartbeat before the group coordinator declares a consumer dead. Since Kafka 3.0 the default is 45000 ms (KIP-735), deliberately raised from the previous 10 seconds so that GC pauses and network jitter don't cause spurious rebalances. The stale recommendation to "raise session.timeout to around 30 seconds" today means halving the default — the opposite of what was intended.

Heartbeats are sent by a separate thread every heartbeat.interval.ms (3000 ms by default), and the "no more than a third of session.timeout.ms" rule is about configuring that parameter, not a description of client behaviour. The upper bound is set by the broker's group.max.session.timeout.ms. The optimum is the smallest value at which consumers aren't ejected from the group in normal operation.

Consumption parallelism

Consumers sharing a group.id split the topic's partitions between them, and each partition is served by exactly one consumer. Maximum parallelism is capped by the partition count: instances beyond that number get no data and act as hot standbys — they aren't idling for nothing, they take partitions over instantly when an active instance fails, which shortens recovery.

Two constraints matter when planning. A topic's partition count cannot be decreased, and increasing it redistributes keys and breaks the per-key ordering guarantee for data already written. That is why headroom is planned up front.

If processing a message is heavy and the bottleneck is your business logic rather than Kafka, adding partitions for parallelism isn't always right. The alternative is to decouple consumption from processing: read with a single consumer and spread the work across a worker pool — including ready-made solutions such as Parallel Consumer — with careful commit management.

Rebalancing

For high-load groups, rebalancing settings affect availability more than any fetch size does.

The eager strategy was the default for a long time: during a rebalance all consumers give up all their partitions and the group stops completely — on a large group that is seconds of downtime on every membership change. Since Kafka 2.4 (KIP-429) a cooperative strategy is available: with partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor, only the partitions that actually change owner are reassigned and the rest keep being processed.

The second most important tool is static membership (KIP-345). Give every instance a unique, stable group.instance.id and a planned restart within session.timeout.ms won't trigger a rebalance at all: the coordinator waits for the same member to come back. For Kubernetes deployments, where restarts are constant, this removes an entire class of rebalance storms.

Kafka 4.0 adds a new broker-side rebalance protocol (KIP-848), enabled with group.protocol=consumer: it removes the stop-the-world phase and moves assignment computation to the coordinator.

If you see lag growing without load growing, start the diagnosis with rebalance-rate-per-hour, not with fetch sizes.

Read isolation and starting position

isolation.level defaults to read_uncommitted, and in that mode a consumer sees records from uncommitted and even aborted transactions. If the system has transactional producers, the consumer must be switched to read_committed — otherwise the entire exactly-once semantics built on the write side is pointless, because the reader gets aborted-transaction data anyway. The price is extra latency: the consumer cannot read past the LSO (last stable offset), that is, past the first uncommitted transaction.

auto.offset.reset (default latest) governs behaviour when there is no stored offset or it has expired. latest means "start from the end", so a new group silently skips everything already accumulated; earliest means "read everything from the beginning", which on a large topic means a read avalanche. none makes the client fail with an error — which for critical pipelines is often preferable to silently skipping data.

Cluster topology: partitions, replication and balancing

A topic's partitions are spread across brokers, each with one leader and RF−1 follower replicas. The leader serves writes and, by default, reads; followers pull data with their own fetch requests. Across multiple zones, replicas of one partition are placed in different zones.

Partition count

Here is the model that often gets described wrongly: a broker does not have a thread per partition. Requests are served by the shared num.io.threads pool and replication by num.replica.fetchers threads. A partition is the unit of parallelism for clients and the unit of data distribution across nodes — but not the unit of thread scheduling on the broker.

That explains the shape of the curve: throughput grows roughly proportionally with partition count only at first, then flattens onto a plateau set by the cluster's disks, network and CPU, and then starts to fall because of overhead. Every partition costs resources — file descriptors for segments and indexes, memory for buffers, its own metadata entries, extra work for fetcher threads.

The common "no more than a few hundred partitions per broker" is a badly outdated figure. In the ZooKeeper era the practical guidance was around 4,000 partitions per broker and about 200,000 per cluster, and the constraint came not from the data but from metadata recovery time and controller re-elections. In KRaft mode — the only mode since Kafka 4.0 — that constraint is gone: metadata lives in a replicated log, and clusters with millions of partitions have been demonstrated publicly.

"As many as fit" is still a poor strategy. Start from the parallelism you need: N parallel processors means at least N partitions, plus reasonable headroom for growth. Headroom matters because partition count cannot be decreased, and increasing it redistributes keys and breaks the per-key ordering guarantee for data already written.

And don't forget to scale num.replica.fetchers along with partitions: growing partitions without growing replication threads is the most common cause of ISR degradation.

Replication factor

RF=3 is the production standard. The cost is concrete: every byte written is shipped to RF−1 additional brokers, so at RF=3 the total disk writes across the cluster and the intra-cluster network traffic are three times what producers send in. That lowers maximum aggregate throughput, especially with acks=all.

Lowering RF for speed is nevertheless not worth it: RF=1 removes fault tolerance entirely, and RF=2 carries a real risk of data loss if one node dies during a rolling upgrade of another. Three brokers minimum and RF=3 is the stable compromise. For secondary topics on clusters with large volumes and modest durability requirements, RF=2 is acceptable.

Separately: the broker's default.replication.factor is 1 by default, so on a new cluster it must be set to 3 explicitly — otherwise auto-created topics end up with no replicas.

If the cluster spans several availability zones, replicas are placed across zones with rack-awareness.

Leadership distribution

Each partition has one leader serving client requests and several follower replicas. Followers pull data from the leader with their own fetch requests, but calling that replication simply "asynchronous" isn't quite right: with acks=all the producer is acknowledged only once the record has been received by every ISR replica, so along the acknowledgement path replication is effectively synchronous.

The load gap between leader and follower is also smaller than people assume: a follower writes exactly the same bytes to disk, and since Kafka 2.4 (KIP-392) it can serve reads as well.

When a topic is created, Kafka distributes partitions round-robin and designates a preferred replica for each — the first in the list. Failures break the balance: when a broker goes down, leadership of its partitions moves to other nodes, and when it comes back it comes back as a follower, so without intervention the skew persists.

Preferred leader election restores the balance by returning leadership to the preferred replicas. The controller can do it automatically: auto.leader.rebalance.enable (default true), with leader.imbalance.check.interval.seconds (300) and leader.imbalance.per.broker.percentage (10%) controlling check frequency and tolerated skew. Manually it is kafka-leader-election.sh with type PREFERRED.

The automatic mode is less obviously right than it looks: a mass leadership transfer itself produces a latency spike and brief NOT_LEADER_OR_FOLLOWER errors for clients. Some operators of busy clusters disable auto.leader.rebalance.enable and rebalance inside a controlled window — including with Cruise Control, which analyses cluster metrics and proposes a redistribution that accounts for the real load on leaders and followers. Which approach you pick depends on what costs you more: rare unplanned latency spikes, or manual control.

Balancing across brokers

Skew appears when a large topic is distributed badly, or when new brokers are added and none of the existing partitions are moved onto them — new nodes don't take data on their own. Distribution can be checked with kafka-topics.sh --describe, broker metrics or Cruise Control, and corrected with kafka-reassign-partitions.sh.

There is a safety condition here that must not be skipped: always run a reassignment with a rate limit. An unthrottled reassign on a loaded cluster saturates network and disks, the ISR collapses, and producers with acks=all start getting errors — it is one of the most reliable ways to take production down with a "planned operation". The limit is set with --throttle (which sets leader.replication.throttled.rate and follower.replication.throttled.rate), and it must be removed afterwards by running with --verify, or the throttle stays and slows normal replication. Start conservative and raise it while watching UnderReplicatedPartitions.

Check the defaults for auto-created topics while you are at it: num.partitions defaults to 1. On a six-broker cluster it makes sense to set num.partitions=6 so new topics spread across all nodes from the start. Better still, turn auto-creation off in production (auto.create.topics.enable=false) and create topics explicitly with deliberate settings.

Don't forget balance inside a node either: with several directories in log.dirs the data is spread across disks, and skew means one disk fills up before the others.

Rack-awareness and follower fetching

If the cluster runs across several availability zones or racks, set broker.rack on every broker. Kafka takes it into account when placing replicas and tries to spread replicas of one partition across zones — so the loss of a whole zone doesn't take out every copy.

The second, frequently underrated setting is reading from the nearest replica (KIP-392, available since 2.4). By default every read is served by the leader, so a consumer in zone A pulls data from a leader in zone B, generating cross-zone traffic. Cloud providers bill that separately, and at scale it becomes a visible line item. Set replica.selector.class=org.apache.kafka.common.replica.RackAwareReplicaSelector on the brokers and client.rack on the consumers, and reads can come from a local follower replica.

One side effect to keep in mind: a follower only serves data up to the high watermark, so such reads can trail reads from the leader by the replication delay.

Reliability and fault-tolerance settings

Minimum in-sync replicas

min.insync.replicas sets the minimum number of replicas, including the leader, that must be in the ISR for the leader to accept a write with acks=all. It is configured at the topic or broker level.

A critically important detail: the default value is 1. Which means acks=all on its own, without explicitly setting min.insync.replicas, protects against nothing — with an ISR shrunk to the leader alone, the write is acknowledged successfully.

On a cluster with RF=3 you set min.insync.replicas=2: then a producer using acks=all is acknowledged only if the message was written to at least two replicas out of three. If fewer are in sync, the write is rejected with NotEnoughReplicasException or NotEnoughReplicasAfterAppendException — and the application is obliged to handle that error rather than treat it as fatal.

The value must be strictly less than RF. Setting it equal to RF (3 of 3) means losing any single replica stops ingest into the topic until it recovers: you trade availability for durability at a ratio almost nobody needs.

Unclean leader election

By default, when a leader fails Kafka only elects a replica from the ISR — one that was in sync with the leader as of the last acknowledged message. That guarantees no acknowledged message is lost. But if no live replica remains in the ISR, Kafka waits for one of the members to return and the partition stays unavailable for both writes and reads.

unclean.leader.election.enable (default false since 0.11) lets you break that rule and elect a lagging replica from outside the ISR. Messages that hadn't replicated are lost, but the partition becomes available without waiting for the downed node.

Where downtime is unacceptable this option is sometimes enabled deliberately, accepting the risk. For critical data it isn't the right move — better to provide enough redundancy and replication throughput that a clean election is always possible.

If it is enabled, monitor UncleanLeaderElectionsPerSec without fail: every firing means data loss actually happened, and that is something to know rather than to learn after the fact from your consumers.

Replica lag

replica.lag.time.max.ms decides how long a follower may fail to catch up before the leader evicts it from the ISR. The default is 30000 ms — raised from 10 seconds in Kafka 2.5 (KIP-537).

The direction of the trade-off here is often described backwards. Evicting laggards quickly increases write availability: the shrunken ISR stops waiting for a slow member and acks=all acknowledgements come faster. What you pay is reduced durability — the message now counts as acknowledged by fewer copies, so the window for losing it on a subsequent leader failure widens. A long timeout does the reverse: it keeps slow replicas in the ISR and preserves the copy count, but with acks=all the producer waits for the slowest member, which hits write p99 directly.

That is why replica.lag.time.max.ms cannot be tuned in isolation from min.insync.replicas: the latter acts as a fuse, stopping the ISR from quietly shrinking to a dangerous level — instead of a silent loss of durability you get an explicit write error.

Too small a value also makes the ISR flap because of brief network delays and GC pauses. In the vast majority of cases the default is right; if you do change it, watch IsrShrinksPerSec and IsrExpandsPerSec — regular shrinks and expansions mean the timeout is badly chosen or replication isn't keeping up (see num.replica.fetchers).

Fsync, page cache and data visibility

Kafka doesn't fsync on every write, relying on background flushing by the OS — which means acknowledged messages exist only in the page cache for a while. Durability here comes from replication, not from the disk.

The visibility mechanics are often described wrongly. A message becomes available to a consumer only once it has been replicated to all ISR replicas and the high watermark has advanced past it. That rule doesn't depend on acks: the parameter decides when the producer gets its answer, not when a reader sees the record. With acks=1 the message doesn't become visible any sooner — lowering acks shortens write acknowledgement, not end-to-end delivery. When a leader fails, records that never crossed the high watermark aren't handed to the new leader and are simply truncated.

Even without forced fsync, replication gives strong durability: loss requires every replica of a partition to fail at once — a whole rack losing power, for instance, which is what rack-awareness works against. Lowering log.flush.interval.ms or log.flush.interval.messages to fsync every message isn't worth it: performance drops sharply, and the benefit is covered instead by adequate RF, min.insync.replicas and healthy replication.

The exception is a single broker with no replication. There is nothing else to provide durability, and frequent flushing is justified — with full understanding of what it costs.

Transactions and exactly-once

Kafka transactions rest on producer idempotence and a transaction coordinator on the broker side. Idempotence is enabled with enable.idempotence=true — the default on 3.0 and later; the producer numbers records and the broker tracks the sequence and drops retries. There is no requirement for max.in.flight=1, contrary to popular belief (covered above in the producer section).

Full transactions additionally need a unique, stable transactional.id on the producer and a sensible transaction.timeout.ms (60000 ms by default), after which the coordinator forcibly aborts a stuck transaction.

Kafka creates the internal __transaction_state and __consumer_offsets topics itself, but with transaction.state.log.replication.factor=3, transaction.state.log.min.isr=2 and offsets.topic.replication.factor=3. That is exactly why transactions don't start on one- or two-broker clusters, and why these values have to be lowered explicitly on dev rigs.

And the thing most often forgotten: transactions on the write side are useless without matching configuration on the read side. The consumer must run with isolation.level=read_committed, or it will see data from aborted transactions and the whole scheme falls apart.

For the end-to-end read-process-write pattern, the source topic's offsets are committed inside the transaction with sendOffsetsToTransaction() — that is what makes "write the result plus advance the offset" atomic.

Transaction overhead isn't as large as commonly believed: for typical pipelines it is single-digit percent of throughput. But it is real, and it grows with short transactions and frequent commits, so the transaction batch size is chosen from measurements rather than intuition.

Monitoring and performance tuning

Kafka exposes metrics over JMX on both brokers and client libraries. In production they are scraped by an exporter — usually Prometheus with the JMX Exporter — and viewed on a dashboard. Without the client half, any change to a producer or consumer config stays unverified.

Client metrics are no less important than broker ones: a large share of the parameters discussed above live in applications, and their effect cannot be checked from the broker side. The metric names below are written the way they actually appear in JMX, so they can be carried straight into an exporter configuration.

Load and latency metrics

Baseline broker throughput figures are in kafka.server:type=BrokerTopicMetrics: MessagesInPerSec, BytesInPerSec, BytesOutPerSec, plus BytesRejectedPerSec and FailedProduceRequestsPerSec for failures. Latency and request rate live in kafka.network:type=RequestMetrics, broken down by type (request=Produce, FetchConsumer, FetchFollower): RequestsPerSec gives the rate, TotalTimeMs the total processing time.

The main diagnostic tool isn't TotalTimeMs itself but its breakdown into phases, available in the same place:

• RequestQueueTimeMs — waiting in the queue before processing
• LocalTimeMs — processing on the leader, including the log write
• RemoteTimeMs — waiting on other replicas, i.e. replication when acks=all
• ThrottleTimeMs — delay caused by quotas
• ResponseQueueTimeMs and ResponseSendTimeMs — queueing and sending the response

This breakdown is what answers "are we slow on disk, in the queue, on replication or on throttling", and you look at the 95th and 99th percentiles, not the averages. RemoteTimeMs rising while LocalTimeMs stays calm points at replication rather than disks, and the fix is num.replica.fetchers, not more I/O threads.

Internal pool load is shown by NetworkProcessorAvgIdlePercent and RequestHandlerAvgIdlePercent (covered in the broker section), and queue sizes by RequestQueueSize and ResponseQueueSize in kafka.network:type=RequestChannel. A permanently full request queue means the broker can't keep up with the flow: either num.io.threads is short, or the disks are the bottleneck. LocalTimeMs tells the two apart.

Replication and lag metrics

UnderReplicatedPartitions (kafka.server:type=ReplicaManager) is the number of partitions with at least one replica behind the leader. It should be zero at all times; a non-zero value means either a broker failure or a follower unable to keep up with replication.

UnderMinIsrPartitions shows partitions where the ISR has fallen below min.insync.replicas. Its appearance is more serious: writes to those partitions with acks=all are already being rejected.

Two more belong in the minimum alert set, from kafka.controller:type=KafkaController. OfflinePartitionsCount — partitions with no leader at all, meaning data is directly unavailable; normal is 0. ActiveControllerCount — summed across all cluster nodes it must be exactly 1: zero means a cluster with no controller, more than one means split-brain.

IsrShrinksPerSec and IsrExpandsPerSec are useful as an indicator of replication instability, and UncleanLeaderElectionsPerSec as an indicator of data loss that has already happened.

On the consumer side the headline figure is lag — how far the group's read position trails the end of the log. It is read with the standard kafka-consumer-groups.sh --describe --group, with exporters such as kafka-exporter, or through Cruise Control. Burrow used to be used for this historically, but the project has been barely maintained for a while and isn't worth building on for new installations. Steadily growing lag means consumers aren't keeping up — either add more of them (and possibly more partitions), or find the bottleneck in the processing itself. A sudden jump usually means a consumer failed or a rebalance happened.

System resources

CPU. Under high load brokers really should be working the processor, especially with TLS enabled and with data being recompressed. If CPU is near 100%, further parameter tuning is nearly pointless and adding nodes is more effective.

Memory. Kafka runs with a relatively small heap — 5–6 GB is the usual figure — leaving as much RAM as possible for the OS file cache, because it is the page cache that lets hot segments be read without touching disk.

Garbage collector. G1GC is the default and the recommendation; for large heaps, ZGC with its sub-millisecond pauses is worth considering. ParallelGC should not be used for brokers, despite the occasional advice to "take it for maximum throughput": it is a fully stop-the-world collector, and its long pauses produce exactly the consequences we are avoiding — the broker misses its exchange with the controller, replicas drop out of the ISR, consumers get ejected from groups. GC pause time and frequency are worth tracking as a signal in their own right.

Disks. Kafka's load profile is sequential writes, so contrary to the common "SSD only" line, a substantial share of large installations run happily on HDDs in a JBOD configuration, which is considerably cheaper at volume. SSD or NVMe genuinely pays off with a large number of partitions (access becomes more random), with heavy catch-up reads of old segments, and where the tail of the latency distribution matters. Watch IOPS, response time and disk queue length, spread data across devices with log.dirs (in KRaft mode JBOD is supported from 3.7, KIP-858), and don't put an excessive number of partitions on a single physical disk.

OS settings, without which none of the above matters: raise the file descriptor limit (ulimit -n of 100,000 and up — running out of them is one of the most common causes of broker failure as partitions and connections grow); set vm.swappiness=1; check vm.max_map_count; raise net.core.rmem_max and net.core.wmem_max when working with large TCP buffers; mount with noatime and prefer XFS.

On KRaft clusters, separately monitor the controller quorum's health and the lag of metadata log replicas.

Client metrics

Everything below is exposed by the clients over JMX and closes the feedback loop: without these numbers, a change to producer or consumer configuration stays an unverified hypothesis.

Producer:

Consumer:

Tuning practice

The first step is deciding what exactly you are optimising: throughput, latency, durability or availability. These are interlinked, and improving one is almost always paid for with another, so trying to "tune everything at once" ends in a configuration that is good at nothing.

For maximum throughput you raise batch and buffer sizes, thread and partition counts, and turn on compression. For minimum latency you do the reverse — lower batch.size and linger.ms, lower fetch.min.bytes. But don't sacrifice acks and idempotence reflexively: measure what they actually cost on your profile first, because it is often less than expected.

The effect of a change is verified with a load test in an environment close to production. The bundled kafka-producer-perf-test.sh and kafka-consumer-perf-test.sh give you baseline numbers quickly with control over message size and target throughput; for more serious scenarios there are Trogdor, which ships with Kafka, and the OpenMessaging Benchmark. The rule is simple: if a change doesn't come with a before-and-after number, it isn't tuning, it's guessing.

Many broker parameters can be changed dynamically through kafka-configs.sh --alter --entity-type brokers — some cluster-wide (--entity-default), some for a single node only (--entity-name <broker.id>). Those that need a restart are rolled one broker at a time, waiting for UnderReplicatedPartitions to return to zero before moving on.

Fully automatic metric-driven tuning sounds appealing but works poorly in practice. First, client parameters such as max.poll.records live in applications, and a broker cannot change them at all. Second, changing configuration automatically on thresholds without hysteresis produces flapping that hurts more than the original problem. The workable version of the same idea isn't autotuning but threshold alerts plus runbooks written in advance: the metric surfaces the problem, a human applies the prepared change.

If the cluster hits its limits even after tuning, move to scaling: add brokers and redistribute partitions — with throttling, always. And if the problem is specifically storage volume, consider tiered storage (KIP-405), which moves cold segments into object storage and keeps only hot data on the brokers.

As load grows, revisit the parameters: settings that were right for 100 MB/s won't be right for 1 GB/s.

Conclusion

The most common cause of failed Kafka tuning isn't a miscalculation, it's outdated advice. acks=1 as the default; max.in.flight=1 to preserve ordering; "raise session.timeout to 30 seconds"; "replication will stall if replica.fetch.max.bytes is smaller than message.max.bytes"; "no more than a few hundred partitions per broker"; "SSD only"; ParallelGC for throughput. Some of it was true once, some was never true, and some is actively harmful today: a max.in.flight above 5 will stop the producer from starting at all.

The second cause is defaults that protect nothing. min.insync.replicas=1 turns acks=all into decoration. default.replication.factor=1 creates topics with no replicas. num.partitions=1 puts a new topic on a single broker. All of it is on out of the box, and none of it gives any warning.

Hence the working order. Check your own version's documentation rather than an article — this one included. Change one parameter at a time and record the before-and-after number: without it, this isn't tuning. And keep the load profile we started with in view — average and maximum message size, target volume in MB/s, required p99, tolerance for loss. A configuration that is optimal for 2 million 200-byte messages will do nothing for a cluster handling 50,000 100 KB messages.