11 - Kafka in Production: Sizing, Retention, Replication and Disaster Recovery

From three test brokers to a cluster that has to stay up

A test cluster with three brokers, default values and no thought-out retention seems to work, until the real load arrives, or a full disk, or a restart done in a hurry. Putting it into production means answering concrete questions: how many brokers and how much disk does my flow need, what do I change compared to the defaults, how do I upgrade the cluster without stopping the writes, how do I rebuild the service if I lose an entire datacenter. This article answers with the method, not with universal numbers: none of the "typical" per-broker throughput values that circulate has a source, and that is why I do not use them. Instead I give you the calculations to do with your own data and the settings verified in the Kafka 4.4 documentation (consulted on 2026-11-12).

The executed material is of two kinds. A Java 21 calculation that sizes network, disk and memory from a load assumption (the figures are declared assumptions, not measurements) and a model that shows why the rolling restart has a golden rule. Everything that requires a cluster, MirrorMaker 2 included, is a template to adapt and was not run.

Sizing is arithmetic on your flow

Start from three numbers that only you know: how many MB per second the producers write, for how many days you want to keep the data, how many consumer groups read everything. From there the quantities that matter follow. Every byte written by a client is written replication factor times in the cluster, once by the client and the others by replication between brokers: the cluster's incoming traffic is write × RF. Outgoing traffic is made of the replicas (write × (RF − 1)) plus the reads of each group (write × groups). The disk is write × seconds of retention × RF, plus free space that you do not want to exhaust. And for memory the documentation offers an explicit estimate: to buffer 30 seconds of activity, write_throughput × 30, because Kafka relies on the operating system's page cache.

import java.util.Locale;

public class Dimensionamento {
    public static void main(String[] args) {
        // ASSUMPTIONS to replace with your own values: none of these numbers is a measurement.
        double scritturaMBs = 50;          // write flow of the clients, in MB/s
        int replicationFactor = 3;
        int gruppiConsumatori = 2;         // each group reads the whole flow
        int giorniRetention = 7;
        double margine = 0.30;             // free space to keep, 30%
        int broker = 3;

        double ingressoCluster = scritturaMBs * replicationFactor;                          // clients + replicas
        double uscitaCluster = scritturaMBs * (replicationFactor - 1) + scritturaMBs * gruppiConsumatori; // replicas + reads
        double discoTotaleTB = scritturaMBs * giorniRetention * 86_400 * replicationFactor / 1_000_000.0; // MB -> TB (decimal)
        double discoConMargineTB = discoTotaleTB / (1 - margine);
        double cache30s = scritturaMBs * 30 / 1000.0;                                       // GB: write_throughput * 30 s

        System.out.printf(Locale.ROOT, "network in, cluster:         %.0f MB/s  (%.0f MB/s per broker with %d brokers)%n",
            ingressoCluster, ingressoCluster / broker, broker);
        System.out.printf(Locale.ROOT, "network out, cluster:        %.0f MB/s  (%.0f MB/s per broker)%n",
            uscitaCluster, uscitaCluster / broker);
        System.out.printf(Locale.ROOT, "total disk (RF %d, %d days): %.2f TB, with margin %.2f TB, per broker %.2f TB%n",
            replicationFactor, giorniRetention, discoTotaleTB, discoConMargineTB, discoConMargineTB / broker);
        System.out.printf(Locale.ROOT, "memory for 30 s of writes: %.1f GB%n", cache30s);

        // Arithmetic check with other data: 100 MB/s, 7 days, RF 3
        double vecchio = 100.0 * 7 * 86_400 * 3 / 1_000_000.0;
        System.out.printf(Locale.ROOT, "check: 100 MB/s x 7 days x RF 3 = %.2f TB (decimal), %.1f TB for 3 brokers%n", vecchio, vecchio / 3);
    }
}

// Output:
// network in, cluster:         150 MB/s  (50 MB/s per broker with 3 brokers)
// network out, cluster:        200 MB/s  (67 MB/s per broker)
// total disk (RF 3, 7 days): 90.72 TB, with margin 129.60 TB, per broker 43.20 TB
// memory for 30 s of writes: 1.5 GB
// check: 100 MB/s x 7 days x RF 3 = 181.44 TB (decimal), 60.5 TB for 3 brokers

With that assumption, three brokers see 50 MB/s incoming and about 67 MB/s outgoing each, 90.72 TB of disk are needed for seven days at replication factor 3, rising to 129.60 TB with the 30% margin, that is 43.20 TB per broker, and 1.5 GB of memory to keep half a minute of writes warm. The number to compare with the real capacity of the network and disks you have is the last of the network figures: if the traffic per broker approaches what the network card or the disks can sustain (a measurement you have to take on your own hardware), you need more brokers. The last line checks the arithmetic with other data: 100 MB/s for seven days at RF 3 is 181.44 TB decimal, and therefore 60.5 TB per broker across three. If you read 173 TB elsewhere for the same case, it is a wrong conversion or one in tebibytes.

For disks the documentation adds that disk throughput is "generally the bottleneck", and that having more disks is better. For partitions the rule seen in the article on the fundamentals holds: you need enough parallelism for the consumers and enough load distribution, bearing in mind that each partition must fit entirely on a single server. There is an official limit on names: the documentation assumes that partitions do not exceed 100,000 per topic, and for that reason topic names cannot exceed 249 characters. In the documentation there are no figures of the type "never more than N partitions per broker".

Broker configuration: little changes, and you know why

Many guides propose long "optimized" server.properties files. The Kafka 4.4 defaults are more reasonable than you might think, and it is best to change them only for a measured reason. Read them before touching them:

• num.network.threads is 3 and num.io.threads is 8; they are raised when the threads turn out to be saturated, not out of habit.
• log.flush.interval.messages defaults to the maximum of a long and log.flush.interval.ms is not set: Kafka leaves the flush to the operating system, and durability is provided by the replicas with acks=all, not by forced flushing.
• min.insync.replicas is 1 and offsets.topic.replication.factor is 3: the first must be raised to 2 for the topics that matter, as in the article on producers.
• controlled.shutdown.enable and auto.leader.rebalance.enable are both true: orderly shutdown and the return of leadership to the preferred replicas are already active.
• auto.create.topics.enable is true: in production, a typo in a topic name creates a new one without anyone noticing, so it is usually turned off.

The list says one useful thing: of the five entries, only two (min.insync.replicas and auto.create.topics.enable) almost always need changing in production, and they concern data safety, not speed. The others are touched when monitoring (see the article on JMX and Prometheus) shows a problem. For the KRaft controllers the documentation estimates that 5 GB of memory and 5 GB of disk for the metadata log are enough for a typical cluster (see the article on KRaft). For the JVM and the filesystem the "Hardware and OS" page gives guidance on file descriptor limits, operating system flush and the choice between XFS and ext4: read it for your system, without copying values from another installation.

A domain topic, and how it is created

The values that define a topic are few: partitions, replication factor, min.insync.replicas and retention. The block below is the creation of a payments topic with the choices discussed; the 30-day retention is a domain choice (how much time you need to reprocess or investigate), not a default. The second command shows increasing the partitions, with the caveat from the operations documentation: if the data is partitioned with hash(key) modulo the number of partitions, increasing them changes the mapping and records with the same key can end up in different partitions; moreover, consumers with auto.offset.reset=latest can lose records produced on the new partitions before discovering them, and the documentation recommends by_duration:<duration>.

# Template to adapt (not run: it needs a cluster reachable at kafka1:9092).
# A domain topic: three replicas, two must be in sync to accept a write with acks=all.
bin/kafka-topics.sh --bootstrap-server kafka1:9092 --create --topic payments-confirmed \
  --partitions 12 --replication-factor 3 \
  --config min.insync.replicas=2 \
  --config retention.ms=2592000000          # 30 days, a domain choice

# Increasing the partitions later is possible, but it moves the keys (see the article on the fundamentals)
bin/kafka-topics.sh --bootstrap-server kafka1:9092 --alter --topic payments-confirmed --partitions 24

# Return leadership to the preferred replicas after restarts (only if auto.leader.rebalance.enable=false)
bin/kafka-leader-election.sh --bootstrap-server kafka1:9092 --election-type preferred --all-topic-partitions

The third command is needed only if you have turned off auto.leader.rebalance.enable: with the default true the cluster returns leadership to the preferred replicas by itself, the behavior described in the "Balancing leadership" section. Never put an unclean election type in a restart procedure: unclean election allows a replica outside the ISR to become leader, and the documentation of the unclean.leader.election.enable parameter (default false) says it can cause data loss.

Restarting brokers without stopping the writes

Upgrading the version, changing a configuration or patching the system requires restarting the brokers. The documented upgrade procedure is: one broker at a time, stop it, upgrade, restart it, and check that the cluster behaves as expected before the next one. Orderly shutdown (controlled.shutdown.enable is true by default) syncs the logs to disk to avoid rebuilding at restart and moves leadership to other replicas before stopping, bringing the unavailability of each partition down to a few milliseconds, as the documentation says. The golden rule is the one everyone writes and few understand: do not restart the next broker until the previous one has rejoined the ISR.

import java.util.*;

public class RollingRestart {
    enum Stato { ATTIVO_IN_ISR, SPENTO, IN_RECUPERO }              // IN_RECUPERO: it is back but has not yet copied the log

    static final int MIN_ISR = 2;

    static long isr(Stato[] broker) { return Arrays.stream(broker).filter(s -> s == Stato.ATTIVO_IN_ISR).count(); }

    static void stampa(String evento, Stato[] b) {
        long n = isr(b);
        System.out.printf("  %-36s ISR=%d  acks=all writes: %s%n", evento, n, n >= MIN_ISR ? "ok" : "REJECTED");
    }

    static void sequenza(String titolo, boolean aspetta) {
        System.out.println(titolo);
        Stato[] b = {Stato.ATTIVO_IN_ISR, Stato.ATTIVO_IN_ISR, Stato.ATTIVO_IN_ISR};   // RF=3, min.insync.replicas=2
        stampa("healthy cluster", b);
        for (int i = 0; i < 2; i++) {
            b[i] = Stato.SPENTO;          stampa("broker " + i + " stopped", b);
            b[i] = Stato.IN_RECUPERO;     stampa("broker " + i + " restarted, copying the log", b);
            if (aspetta) { b[i] = Stato.ATTIVO_IN_ISR; stampa("broker " + i + " back in the ISR", b); }
        }
    }

    public static void main(String[] args) {
        sequenza("restart that waits for the full ISR:", true);
        sequenza("restart without waiting:", false);
    }
}

// Output:
// restart that waits for the full ISR:
//   healthy cluster                      ISR=3  acks=all writes: ok
//   broker 0 stopped                     ISR=2  acks=all writes: ok
//   broker 0 restarted, copying the log  ISR=2  acks=all writes: ok
//   broker 0 back in the ISR             ISR=3  acks=all writes: ok
//   broker 1 stopped                     ISR=2  acks=all writes: ok
//   broker 1 restarted, copying the log  ISR=2  acks=all writes: ok
//   broker 1 back in the ISR             ISR=3  acks=all writes: ok
// restart without waiting:
//   healthy cluster                      ISR=3  acks=all writes: ok
//   broker 0 stopped                     ISR=2  acks=all writes: ok
//   broker 0 restarted, copying the log  ISR=2  acks=all writes: ok
//   broker 1 stopped                     ISR=1  acks=all writes: REJECTED
//   broker 1 restarted, copying the log  ISR=1  acks=all writes: REJECTED

In the orderly sequence, when broker 0 has been restarted it does not yet count in the ISR: the cluster stays at two in-sync replicas, the minimum, and keeps accepting writes; only after it rejoins do you move on to broker 1. In the hasty sequence broker 0 is back but is still copying the log; stopping 1 leaves a single in-sync replica, below the minimum, and all the writes with acks=all are rejected. This is the mistake the rule prevents, and you see it in monitoring as UnderMinIsrPartitionCount above zero. In the model the "states" are three and the passage from IN_RECUPERO (recovering) to rejoined is instantaneous if you wait; in a real cluster it depends on the amount of data to be copied back, and that is why there is no correct fixed wait: you wait for the signal, not for the time.

Disaster recovery: MirrorMaker 2

Replicas inside a cluster protect against a broker failure, not against the loss of a datacenter. For that you need a second cluster in another location and a process that copies the data: MirrorMaker 2, built on Kafka Connect, described by the documentation as a geo-replication tool. You declare the clusters, the replication flows (primary->secondary.enabled), the topics to copy (regular expressions) and the consumer groups whose state is to be replicated. The copied topics change name: by default, with the DefaultReplicationPolicy, the topic orders-placed of the cluster primary becomes primary.orders-placed in the secondary, so as not to mix records of different origins.

# Template to adapt (not run): active/passive replication, from the Kafka 4.4 documentation example.
clusters = primary, secondary
primary.bootstrap.servers = broker3-primary:9092
secondary.bootstrap.servers = broker5-secondary:9092

# The flow goes in one direction only: from the primary to the secondary
primary->secondary.enabled = true
primary->secondary.topics = orders-.*, payments-.*
primary->secondary.groups = service-.*

# Launched in the secondary's datacenter ("consume from remote, produce to local"):
#   bin/connect-mirror-maker.sh connect-mirror-maker.properties --clusters secondary

Three details, all from the documentation. First, the recommendation "consume from remote, produce to local": the process is launched close to the destination cluster (--clusters secondary), because producers suffer from slow networks more than consumers do. Second, MirrorMaker processes share configuration through the destination cluster, so two processes with different topic lists towards the same cluster conflict: keep a single configuration file. Third, exactly-once is supported for dedicated MirrorMaker clusters from version 3.5.0, but it must be enabled (exactly.once.source.support) and it requires internal communication between the nodes.

The most delicate problem remains: offsets. A record has one offset in the primary and another in the secondary, so a consumer that switches to the secondary cannot reuse its own offsets. The way to translate them is in KIP-382: the RemoteClusterUtils class exposes translateOffsets(properties, targetClusterAlias, consumerGroupId, timeout), which finds the local offsets corresponding to the last checkpoint of the origin group, assuming the DefaultReplicationPolicy. "Last checkpoint" means that the translation has the resolution of the checkpoints: after a failover some records can be read again, so the consumer must tolerate duplicates (see the article on idempotency). I did not run a failover: the procedure must be tried on two test clusters before you trust it.

What you can do now

You can compute the network, disk and memory of a cluster from your own data, and you recognize a calculation that is wrong by a unit factor. You know which broker settings are really worth changing and which are already at the right value. You can create a topic with min.insync.replicas and retention chosen for domain reasons, and you know that increasing the partitions moves the keys. You can restart brokers one at a time waiting for the ISR, and you know why never an unclean election. You know that MirrorMaker 2 copies topics with a prefix and translates offsets with checkpoints, but you also know that a failover is rehearsed first. If a message will not let itself be processed after all this, the answer is the dead letter queue.