DEV Community: Timofei Ivankov

Configuring Apache Kafka for High-Load Systems

Timofei Ivankov — Tue, 18 Aug 2026 12:20:26 +0000

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:

Metric	What it tells you
`batch-size-avg`	actual batch size against `batch.size`
`record-queue-time-avg`	the effect of `linger.ms`
`buffer-available-bytes`, `waiting-threads`	whether the app is hitting `buffer.memory`
`request-latency-avg`	latency of a request to the broker
`record-error-rate`, `record-retry-rate`	whether the producer is living on retries
`compression-rate-avg`	whether compression is paying off

Consumer:

Metric	What it tells you
`records-lag-max`	lag on the worst partition, the headline figure
`fetch-latency-avg`, `fetch-size-avg`	the effect of `fetch.min.bytes` and `fetch.max.wait.ms`
`records-per-request-avg`	how full fetches are
`commit-latency-avg`	the cost of commits
`time-between-poll-avg`	how close you are to `max.poll.interval.ms`
`rebalance-rate-per-hour`	group stability

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.

Key Microservice Architecture Patterns: Strangler Fig, API Gateway, Service Mesh, and More

Timofei Ivankov — Sat, 15 Aug 2026 18:25:30 +0000

Five services synchronously calling each other, each with 99.9% availability, yield 99.5%. Splitting a monolith doesn't add reliability in itself — it takes it away, and you have to restore it separately.

Why split a monolith then? Independent deployment, separate scaling of the high-load parts, boundaries of responsibility between teams. The patterns below are about how to get that and what you pay for it. Strangler Fig and API Gateway, Service Mesh and Sidecar, Database per Service, CQRS, Event Sourcing.

Strangler Fig

Strangler Fig is a way to rewrite a monolith without ever stopping it: a facade is placed in front of it, and functionality moves out of the monolith into new services piece by piece until nothing is left. The name was coined by Martin Fowler — a strangler fig grows into the fork of another tree and gradually takes its place.

Three stages. The facade takes all traffic from day one and gradually moves it from the monolith to the microservices: the monolith's share drops until it stops receiving requests at all. After the migration the facade is usually not thrown away — it stays as the system's entry point.

Where to start. The first piece you extract isn't the most important one, it's the most convenient: few incoming dependencies, its own data boundary, noticeable but not critical traffic. Usually that's something at the edge of the system — notifications, report generation, search, export. The domain core, where all the transactions converge, is a bad first candidate: you can't extract it without untangling all the data at once, and that's exactly where migrations tend to stall. The point of the first step isn't to get a valuable service, it's to have the team walk the whole path once — extraction, deployment, monitoring, rollback — on something they can afford to break.

Where to draw the boundary. By domain contexts, and there's a single test for a correct one: the business operation fits entirely inside one service and doesn't require a distributed transaction. If placing an order means synchronously calling three new services, the boundary is wrong — you've reproduced the couplings you had in the monolith and added a network between them.

What to do about the data. Forwarding requests is the easy part. The hard part is that the extracted service needs data that still lives in the monolith's database, and for the duration of the migration — months — someone has to own it. There are three options, and you need to pick one before the first traffic switch.

The new service reads the monolith's database. Fast to start, but you get a distributed monolith: two applications coupled through one database schema, and the "shrinking" usually stops right there. A temporary measure for one release cycle, no longer.
Data is copied into the new service's database, the source of truth stays in the monolith. Synchronization is one-way — through events or by reading the transaction log (CDC). The new service can only read; all changes still go through the monolith. This is a working intermediate state you can live in for a long time.
Ownership is handed to the new service, and the monolith calls it through an API. The final state. Getting there requires a short dual-write window and a reconciliation afterwards — the most delicate operation in the whole migration, and one you do for a single service at a time.

Rollback has a limit worth understanding. While the new service only reads, moving traffic back to the monolith is a route switch in the facade, a matter of seconds. Once the service owns the data, rollback in that sense no longer exists: it has accumulated changes that don't exist in the monolith, and "moving the load back" means a reverse data migration, not a config edit. So the ownership handover is planned separately — with a window, a reconciliation, and a return procedure written in advance.

When the migration is done — and why it often isn't. The pattern has a characteristic failure that gets discussed less than the pattern itself: in the first year 60% of the functionality moves out, the remaining 40% turns out to be the most entangled, budget and interest run out — and the company is left forever with a monolith, microservices, and a facade in between, meaning both systems and the cost of operating both. This isn't a hypothetical risk, it's the most likely outcome if nobody watches for it.

The antidote is organizational: from day one, measure the share of what's still in the monolith — by endpoints, by traffic, by tables, it doesn't matter as long as it's always the same way — and require that number to move every quarter. If it hasn't changed in a quarter, the migration has stopped, and that calls for a decision. "We're stopping here deliberately, the rest of the monolith stays forever, and so does the facade" is a perfectly acceptable decision, as long as it's made explicitly.

The facade isn't a free component. All of the system's traffic goes through it from day one, which makes it a single point of failure with everything that follows: fault tolerance, clustering, monitoring, a dedicated owner. The same requirements are covered below in the API Gateway section — and that's not a coincidence: the Strangler Fig facade is usually the system's future API gateway, just early in its life. It's worth choosing it with that in mind rather than standing up a temporary proxy you'll have to replace.

API Gateway

The API gateway is the only address the clients know. It takes an external request, decides which service should handle it, and returns the response; the internal topology is invisible from outside.

Clients talk to a single gateway instead of the services directly. It terminates external HTTPS, handles authentication, rate limiting, caching and logging, and then calls the internal services over their own protocols — REST or gRPC. The external contract doesn't depend on how the internal network is arranged.

The gateway can fan a single external request out into several internal ones and assemble the response: an order page is put together from the orders, delivery and reviews services in one client call instead of three.

But only reads can be assembled this way. An operation that changes state across several services — placing an order together with charging the payment — can't be split the same way: if the charge went through and the order service didn't answer, the gateway has nothing to roll back with, no shared transaction and no compensations. You're left with "money charged, no order." Distributed changes are Saga's job, not the gateway's.

Beyond routing, a gateway usually takes on cross-cutting concerns: TLS termination, logging, caching, rate limiting, protocol translation from external REST into internal gRPC. One of them deserves a separate note.

Authentication at the gateway doesn't mean nobody checks further down. The gateway authenticates the external caller and turns its token into an internal context, but the services themselves are obliged to verify that context. Otherwise anyone who ends up inside the perimeter — a neighbouring service, a compromised pod, a contractor on the same network — gets the privileges of any user.

How much of this you write yourself depends on what's already there. In Kubernetes, external routing and balancing between replicas are already covered by the platform, and a separate gateway is only needed for aggregation, public API versioning and managing keys for external consumers. One idea worth borrowing from there regardless: the platform team owns the entry point, the service teams own their routes. Otherwise the shared configuration quickly becomes nobody's.

Pick the versioning scheme up front. There are three. A version in the path (/v2/orders) is the easiest to debug and cache, but the version leaks into the resource address. A version in a header keeps the address clean, but it's invisible in logs and the request is harder to reproduce by hand. Content negotiation via Accept is formally the most correct and the rarest in practice. Most public APIs take the first.

The cost of a gateway isn't only the single point of failure. That the whole system becomes unreachable when the gateway goes down is obvious — hence clustering, a dedicated owner, and monitoring that isn't tied to the services behind it. Throughput is the less obvious part. On a synthetic setup where an nginx gateway and the service run in containers on the same machine — so there's no network between them at all — proxying added roughly 0.4 ms per request, and throughput dropped threefold: from ~220k to ~75k requests per second. Latency is rarely the critical part; a threefold difference means you'll have to scale the gateway before the services behind it.

Backend for Frontend is the variation where each client type gets its own gateway. You introduce it once the data sets for web and mobile have genuinely diverged. The price is that cross-cutting logic — authorization, rate limiting, logging — gets duplicated across gateways and starts drifting apart over time, which is why BFFs are created in response to actual divergence rather than pre-emptively for each platform.

Service Mesh

A service mesh and an API gateway aren't competitors, they're different traffic axes, and that's worth separating right away. The gateway handles north-south: entry from outside, external clients, their authentication, the public contract. The mesh handles east-west: calls between your own services inside the perimeter, where there are no external clients. A large system runs both, and a request path looks like this: client → gateway → service A's proxy → service B's proxy → service B. Istio can also cover north-south with its ingress gateway, which is why on plain Kubernetes the two layers sometimes collapse into one product.

Two traffic axes. The API gateway owns north-south — entry from outside. The service mesh owns east-west: a proxy next to each service intercepts the calls, encrypts them between proxies over mTLS, and gets its configuration from the control plane. The services themselves talk to their own proxy over localhost and don't know the mesh exists.

A mesh is built as a distributed system of proxies. The data plane is a proxy next to each service intercepting all inbound and outbound traffic; the services "think" they're talking directly to each other. The control plane hands them configuration, issues and rotates certificates, and collects telemetry. Rules like "10% of traffic to v2" or "turn on mTLS everywhere" are set through it — but not by hand: they're described as Kubernetes resources, kept in a repository and delivered through CI, otherwise you lose the main thing, reproducibility.

Since 2024 Istio has a second model, ambient: one shared ztunnel proxy per node covering L4 — mutual TLS, authorization, metrics — with an L7 proxy brought up only where retries, traffic splitting and HTTP parsing are actually needed.

Because all traffic goes through a proxy, a mesh can do whatever a proxy can: balance, split traffic across versions for canaries, retry failed calls, trip the circuit, record the latency and status of every call. None of it requires touching the services.

But that's usually not why a mesh gets adopted. There's normally a single reason: the requirement to encrypt traffic inside the perimeter comes from security or a regulator, and a mesh satisfies it without changing the applications — the proxies bring up mTLS between themselves and the services never know.

There's one thing a mesh doesn't cover, though people expect it to. It forwards tracing headers between services, but it can't carry them through your code: if a service accepts a request and moves on without copying those headers into its outgoing call, the chain breaks. Free tracing doesn't exist; a minimal change to the applications is required anyway.

The overhead is worth knowing before you adopt it. By Istio's own measurements, at 1,000 requests per second with a 1 KB payload and mTLS enabled, a single sidecar proxy costs 0.20 vCPU and 60 MB of memory — per pod. Across two hundred pods that's 40 vCPU and 12 GB spent on infrastructure alone. Ambient in the same configuration costs 0.06 vCPU and 12 MB per node.

A misconfigured control plane breaks communication across the entire system at once rather than in a single service, and nobody except the people who set the mesh up will be able to sort it out. A practical guideline: fewer than two dozen services and nobody demanding encryption inside the perimeter — an API gateway plus a retry library with timeouts will do.

Sidecar

A sidecar is a helper process living next to the application and taking on everything that isn't business logic: outbound TLS, shipping logs, fetching configuration. In Kubernetes it's a second container in the same pod, and the two talk over localhost.

A sidecar runs in the same pod as the application and shares its lifecycle. The application talks to it over localhost in the clear; outbound, the sidecar speaks HTTPS and is the one responsible for certificate verification, retries and log shipping. The business logic knows nothing about TLS or the log server.

A sidecar most often plays one of two roles, and both have names. Ambassador faces outward: it represents the application on the external network and takes on TLS, retries, protocol translation — the application sends plain HTTP to localhost and the sidecar establishes HTTPS. Adapter faces inward: it converts what the application emits into the format the platform expects. The classic example is a metrics exporter — the application publishes metrics its own way and the sidecar turns them into something Prometheus can read.

What gets reused is the image, not the instance. The same image is attached to different applications, but every pod always has its own copy running.

A sidecar's independence is easy to overestimate. Its behaviour genuinely does change through configuration without rebuilding the application — a new retry policy or a new log server address doesn't touch the code. But you can't update the sidecar's image without touching the application: in Kubernetes any change to the pod spec, including swapping the image of one container, recreates the whole pod. You can't scale it separately either — a sidecar lives one-to-one with an application instance by definition and is addressed over localhost.

A sidecar has to start before the application and stop after it. Otherwise the application loses its first requests at startup and its last ones at shutdown. Ordinary containers in a pod give you no such guarantee: their start and stop order is undefined, and the race where the application comes up before the proxy is a classic problem in early service mesh rollouts. The correct mechanism is to declare the sidecar in initContainers with restartPolicy: Always — then the kubelet guarantees it is running before the main container starts and stopped after it exits.

And limits. A memory limit set too low kills not the sidecar but the pod's entire network path: the application stays alive and simply stops being able to reach anything.

Here's the example where this kind of sidecar most often breaks. A Python application makes an ordinary request:

response = requests.get("http://localhost:8080/api/data")

And next to it sits nginx, taking that in the clear and going out over HTTPS:

location / {
    proxy_pass https://external-service:443;
    proxy_ssl_server_name on;        # off by default — SNI is not sent
    proxy_ssl_verify on;             # off by default — the certificate is not checked
    proxy_ssl_trusted_certificate /etc/ssl/certs/ca-certificates.crt;
    proxy_ssl_verify_depth 2;        # 1 by default
    proxy_ssl_protocols TLSv1.2 TLSv1.3;
}

The first two directives are mandatory, and both are off by default. Without proxy_ssl_server_name on the sidecar doesn't send SNI, and any host behind a CDN or behind a load balancer serving several certificates will abort the handshake: the application gets a 502 and the log shows SSL_do_handshake() failed ... tlsv1 unrecognized name. Without proxy_ssl_verify on the sidecar doesn't check the external service's certificate at all — it will accept a self-signed one issued to somebody else's name and hand the application a perfectly honest 200.

Hence a rule worth keeping in mind whenever you use a sidecar: moving a function out of the application and into a sidecar means you inherit the sidecar's defaults, not the defaults of the library you used before. Python's requests verifies certificates by default; nginx as a reverse proxy does not. Naively moving TLS into a sidecar doesn't raise security, it lowers it — and there's no way to notice without a deliberate check, because from the outside everything looks like it's working.

Service mesh proxies are sidecars, so everything above about start-up order and limits applies to them first of all.

Database per Service

Every service has its own database and never touches anyone else's directly — only through the owner's API. This isn't about a physically separate server: several services can live inside one PostgreSQL instance in separate schemas their neighbours can't reach. What matters isn't hardware isolation, it's that you can change the schema without asking anyone.

Each service owns its database and is the only one that touches it. Services exchange data through their APIs, and reaching directly into someone else's database is forbidden: that's exactly what turns a set of microservices into a distributed monolith where you can't change a schema without breaking a neighbour.

One expectation is worth correcting straight away. Splitting the databases doesn't isolate failures by itself: if service A calls B synchronously and without a timeout, B's failed database will take A down with it — first the worker threads run out, then the connections, and the failure travels up the call chain. Isolation appears when the calling side has timeouts and a circuit breaker; splitting the databases merely makes it possible.

Where to draw the boundary. Local transactions only pay off if the business operation fits entirely inside the service. The criterion: everything that must change together and be checked for consistency at the moment of the change has to end up inside one service. If enforcing "you can't order an item that isn't in stock" requires a synchronous look into somebody else's database, the boundary is wrong — and from there you'll either break isolation or build a saga where there needn't have been one.

You can't JOIN across services. The data you need is either fetched over an API or kept locally as a projection the service maintains by listening to the owner's events. And here's the trap almost everybody falls into: if the owner first writes the change into its database and then publishes an event in a separate call, that's a write into two systems without a shared transaction. If the broker goes down between the two calls, the change exists and the event doesn't — subscribers never learn about it, and no error is raised anywhere. The fix is Transactional Outbox: the event is written into the same database, in the same transaction as the change itself, into a separate table, and a separate process reads that table and publishes to the broker.

Transactions are local now. Consistency between services becomes eventual, achieved through events or sagas. A saga breaks a business operation into a chain of local transactions, each publishing an event that triggers the next step. There's no rollback in the usual sense: instead, every step gets a compensating action. Not "undo the charge" but "issue a refund" — a compensation is an ordinary business operation, visible in the history, not a rollback that erases its own traces.

Connections run out before the database's resources do. This is the limit you actually hit in practice, and it isn't obvious. Every replica holds its own pool and the totals multiply: twenty replicas with a pool of ten each is two hundred connections to one database. PostgreSQL defaults to max_connections = 100, three of which are reserved for the superuser — so you'll hit the ceiling long before CPU or memory run out. The fix isn't raising max_connections (every connection costs memory and loads the planner), it's an external pooler such as PgBouncer in transaction pooling mode.

Reports that cross service boundaries are not built with API composition or nightly scripts but with CDC: a separate process reads each database's transaction log and streams the changes into an analytical store, asking nothing of the services themselves. The best-known implementation is Debezium. This is a separate architectural layer and it's better planned in advance.

One detail worth calling out: the orders service stores the item price as of the order rather than fetching it from the catalogue. That's not accidental duplication, it's a deliberate copy — for the order it's part of its own history and must not change retroactively when the price list is updated.

One database shared by several services isn't a mistake, it's a trade-off with a clear price. You get familiar ACID transactions across service boundaries and one database to operate, and you pay with schema coupling (changing a table requires coordination between teams) and runtime coupling (a long transaction in one service blocks another). In the microservice pattern catalogue, Shared Database is listed as a pattern with a list of trade-offs, not as an anti-pattern. The problem isn't that you can't do it, it's that getting out later costs more than not getting in: the longer a shared schema lives, the more code grows into it. So if a shared database is a deliberate choice, write down up front the condition under which you'll leave it.

CQRS

CQRS splits writes and reads: a command changes state and returns next to nothing, a query reads and changes nothing, and the two use different models.

A command changes state through the write model; a query reads a prepared representation from the read model. In the simple form both sides work against one database; in the advanced form they use separate stores and the read store is updated from events with a lag.

Martin Fowler wrote a separate note on the pattern, and his position is worth quoting accurately, because it's often reported backwards: he is cautious about CQRS and states plainly that it should be used with considerable care, that the domains it suits are a clear minority, and that it belongs to a single bounded context rather than to a system as a whole.

The pattern has two forms, and that distinction matters more than anything else. The simple form is one database but different models in code: a command model with invariants and checks, a query model with denormalized representations shaped for specific screens. The advanced form is separate stores: one takes the writes, the other is updated from events and serves the reads.

What follows — both in this article and in most other material — deals almost exclusively with the advanced form, which leaves the impression that CQRS necessarily means two databases and an event bus. It doesn't. The vast majority of systems are fine with the simple form: it delivers most of the benefit and drags along neither eventual consistency, nor projections, nor separate infrastructure. Separate stores get introduced when you've hit a measured limit — read replicas can no longer keep up, or writes and reads need fundamentally different guarantees. "We did it properly from the start" is a poor reason for a second database.

Once the stores are separate, a lag appears: after a command the change isn't immediately visible on the read side. There aren't many ways to live with that. A client-generated identifier lets you display the created entity right away without waiting for the projection. A command can return a version number, and a subsequent read waits until the projection catches up to it — that's read-your-writes. Finally, reads from the author of the change can be routed to the write side for a few seconds while everyone else reads from the read side. What you shouldn't do is ship an interface where the user hits "Save" and doesn't see their own data on the next screen.

Sooner or later a projection will need rebuilding — a bug in a handler, a lost event, a changed requirement. It's a routine operation and it has to be planned for. With Event Sourcing the answer is simple: wipe the projection and replay the log from the beginning. Without Event Sourcing there's no log, so you rebuild from the write model — which needs a process able to read the current state of the write database and reassemble the read tables, plus a way to tell that the projection has fallen behind or diverged: a counter of processed events and a regular reconciliation. Without such a process, the first bug in a projection gets fixed by hand, at night.

// Command model — handler for the create-order command
public class OrderCommandHandler {
    private final OrderRepository repo;
    private final OutboxRepository outbox;
    private final TransactionTemplate tx;

    public OrderCommandHandler(OrderRepository repo, OutboxRepository outbox, TransactionTemplate tx) {
        this.repo = repo;
        this.outbox = outbox;
        this.tx = tx;
    }

    public void handle(CreateOrderCommand cmd) {
        // 1. Validate the business rules
        // 2. Build the order, add items, compute the total
        Order order = new Order(cmd.getOrderId());
        // ... (add items and the rest)
        // 3. The order and the event are saved in ONE transaction to ONE database
        tx.execute(() -> {
            repo.save(order);
            outbox.save(new OrderCreatedEvent(cmd.getOrderId()));
        });
        // 4. A separate process reads the outbox table and publishes to the broker
    }
}

// Query model — handler for the get-order query
public class OrderQueryService {
    private final OrderViewRepository readDb;

    public OrderQueryService(OrderViewRepository readDb) {
        this.readDb = readDb;
    }

    public OrderDto handle(GetOrderQuery query) {
        // Point lookup by key against the denormalized read-model table
        return readDb.findById(query.getOrderId())
                     .map(OrderDto::from)
                     .orElse(null);
    }
}

Two details here answer questions that come up immediately.

The order identifier arrives in the command rather than being issued by the database. That's the answer to "how does the client learn the id if the command returns nothing?" — the client generates it, usually as a UUID. A side benefit is that the command becomes idempotent: sending it again with the same identifier won't create a second order.

The order and the event are saved in one transaction to one database. The naive version — repo.save(order) followed by eventBus.publish(...) — is the same dual write from the Database per Service section: if the broker is down between the two calls, the read model never learns about the order. Hence the outbox table and the separate process that drains it.

Splitting write and read across separate services is possible but not required — CQRS lives happily inside a single one. Splitting into services makes sense when the two sides have diverging scaling requirements, not for its own sake.

Event Sourcing

State isn't stored, it's computed: events go into a log, and an aggregate's current state is obtained by applying them in order. state = f(all past events). That's also where the ability to get the state at any point in the past comes from — replay the log up to that point.

State isn't stored: events are appended to the log, and an aggregate's current state is assembled by applying them in order. A snapshot is a cached point you can start replaying from instead of starting at zero.

The obvious benefit is a full history and audit trail, but replay has a caveat people notice late. It only answers new questions with data the events already carry. If MoneyDeposited was saved without a timestamp or without the channel of the operation, no new logic will recover them: replay reproduces history, it doesn't invent it. Hence the rule for designing events — record everything you know at the moment the event occurs, even if nobody needs it today. Disk space is cheaper than being unable to answer a question two years from now.

Events live in an event store, grouped by aggregate: every event for order 1234 is one sequence. To get the state you load them and apply them to an empty object:

Order order = new Order();
for (Event event : eventsForOrder1234) {
    order.apply(event);
}

Everything hinges on apply. It's tempting to declare one overload per event type — apply(OrderCreated), apply(ItemAddedToOrder) — but that won't compile: overloads in Java are picked by the static type of the argument, and inside the loop it's always Event. So there's a single apply with the dispatch inside:

public void apply(Event event) {
    switch (event) {
        case OrderCreated e     -> { this.id = e.orderId(); this.status = "created"; }
        case ItemAddedToOrder e -> { items.add(e.item()); total = total.add(e.price()); }
        case OrderShipped e     -> { this.status = "shipped"; this.shippedAt = e.date(); }
        default -> throw new IllegalStateException(
                "Unknown event type: " + event.getClass().getName());
    }
}

The default branch is mandatory, and it isn't paranoia. Six months from now a new event type shows up — say MoneyTransferredOut. Without that branch, reconstruction silently skips it: no exception, nothing in the logs, and the state comes out wrong by exactly the sum of all such events, which surfaces during a reconciliation an unknown amount of time later. Reconstruction from the log should fail on an unknown event rather than count its way to a plausible but incorrect number.

Snapshots are a periodically saved state so you don't replay a long history from scratch. Two caveats, both practical. They're needed by a minority of aggregates: if a typical aggregate has dozens of events, replaying costs a fraction of a millisecond, while a snapshot adds code and one more place where data can diverge — introduce them based on measurement. And a snapshot is a cached result of apply, so changing the logic of apply invalidates every existing snapshot: version them and be able to discard them wholesale.

Appending an event isn't just an append, and without this the whole construction is unsafe. If two processes modify the same aggregate at once, both read the same history, both decide on the basis of it, and both append their events — you end up with two debits against a balance that was only enough for one. The protection is optimistic locking by version: on read you remember the number of the last event, on write you pass it as the expected one, and the insert goes through a unique index on the (aggregate_id, version) pair. If somebody wrote in the meantime, the insert violates uniqueness, the command fails, and it has to be retried after re-reading the history.

You can't correct what's written. If an event is wrong, a compensating one is appended and the wrong one stays in the log. You can see what was recorded, when it was corrected, and by how much.

Personal data in an immutable log is a limitation people remember late, and it sits awkwardly with the very domains where Event Sourcing is recommended first. In an append-only log, a subject's request to delete their personal data (GDPR) can't be satisfied: an event can't be modified or deleted without breaking the model. The standard workaround is crypto-shredding: personal fields are written into the event encrypted, the key is stored separately and tied to the subject, and deleting the subject means deleting the key. The events stay put, the history is intact, and the personal fields can no longer be read. This has to go into the event schema from the very beginning — retrofitting encryption into an existing log is considerably more expensive.

An endlessly growing stream for a single aggregate usually means the aggregate boundary was drawn wrong. A "Product" aggregate with a million events over five years isn't a case for snapshots, it's a case for revisiting the model. Healthy aggregates have a natural end: an order closes, an account closes, a reporting period closes — and completed streams can be moved to an archive without touching the active ones.

Event schemas evolve painfully: you can't just change a field, the old events are already in the log. You either version the events or migrate the log, and migrating a log is far from trivial.

There are two kinds of events and mixing them is a mistake. The ones in the event store are internal, domain events — designed for reconstructing aggregate state and a private detail of the service. What gets published outward are separate integration events: coarser, more stable, with a versioned schema. Publish the internal ones directly and every subscriber becomes coupled to your aggregate's internal model, at which point event versioning stops being an internal concern of the service and turns into release coordination between teams.

Conclusion

What all seven have in common isn't the problems they solve, it's the exact place where they defy expectation. Splitting databases doesn't isolate failures — timeouts and a circuit breaker on the calling side do. Authentication at the gateway doesn't mean the services can skip checking. A sidecar can't be updated or scaled separately from its application. CQRS doesn't require two databases. A mesh doesn't give you free tracing. In Strangler Fig, rollback exists right up until data ownership changes hands. Every one of these corrections costs more than the pattern itself, because you find it in production.

The second thing they have in common is that almost all of them add a component that needs an owner. A facade, a gateway, a mesh control plane, a sidecar in every pod, a process draining the outbox, a process rebuilding projections. That's not a box on an architecture diagram — it's an on-call rotation, upgrades, and its own page in the runbook. Before adopting a pattern it's worth asking who fixes it at three in the morning.

You'll end up combining them anyway: a Strangler Fig migration almost always comes with splitting the databases, and CQRS doesn't work without an outbox. But picking them up "to do it properly" is the worst possible reason. Each has a measurable condition under which it pays for itself, and in most cases that condition hasn't arrived yet.

Best practices for creating fault-tolerant systems

Timofei Ivankov — Sat, 15 Aug 2026 12:55:41 +0000

An external service started responding in 3 seconds instead of 30 milliseconds. Nothing crashed, the logs are clean, but the connection pool is full, the request queue is growing, and a minute later your service is down together with it — even though it is perfectly healthy.

Fault tolerance is a set of decisions about what to do at moments like this: whether to retry, how long to wait, what to return when there will be no answer at all. Almost all of them are made in code, not in the cluster configuration. Below is how they look in Python and Java, on ordinary tasks: calling an external API, querying a database, background processing.

Retry and exponential backoff

When a call to an external service fails over the network, it is reasonable to try again — the failure may have lasted a second. Everything else about retries is not obvious.

Python, Tenacity

Retries are easy to write by hand: a while, an attempt counter and time.sleep(). The problem is that a handwritten loop is almost always naive: a fixed pause with no spread, except Exception instead of telling error types apart, no budget, no ceiling on total time. Tenacity turns those decisions into explicit parameters: you have to make them consciously instead of skipping them by default.

import logging
import requests
from tenacity import (
    retry,
    stop_after_attempt,
    wait_random_exponential,
    retry_if_exception,
    before_sleep_log,
)

logger = logging.getLogger(__name__)


def is_retryable(exc: BaseException) -> bool:
    if isinstance(exc, (requests.exceptions.ConnectionError,
                        requests.exceptions.Timeout)):
        return True
    if isinstance(exc, requests.exceptions.HTTPError):
        code = exc.response.status_code if exc.response is not None else None
        # 429 - we are being asked to wait, 5xx - the server broke.
        # Retrying 4xx other than 429 is pointless: the request won't get any more valid.
        return code == 429 or (code is not None and 500 <= code < 600)
    return False


@retry(
    stop=stop_after_attempt(3),
    # full jitter: without a random spread every client
    # comes back in the very same millisecond
    wait=wait_random_exponential(multiplier=0.1, max=5),
    retry=retry_if_exception(is_retryable),
    # without reraise the caller gets tenacity.RetryError
    # instead of the original requests exception
    reraise=True,
    before_sleep=before_sleep_log(logger, logging.WARNING),
)
def fetch_data(url):
    # the timeout is mandatory: without it retries hang instead of retrying
    response = requests.get(url, timeout=(1, 3))
    response.raise_for_status()
    return response.json()


try:
    data = fetch_data("https://api.example.com/data")
except requests.exceptions.RequestException as e:
    logger.error("Failed to fetch data: %s", e)

The @retry decorator does not catch every exception, only those that pass the is_retryable check: network errors, timeouts, 429 and 5xx. A 400 Bad Request goes up on the very first attempt — retrying it is pointless, the request will not become any more valid.

The pause grows exponentially, but what is taken is not the bound itself: it is a random value between zero and that bound (the ceiling goes 0.1 → 0.2 → 0.4 … up to 5 seconds). This is full jitter. If a service went down under a thousand clients, then without a spread all of them come back at once and take it down again at the exact moment it was coming back up.

reraise=True is not cosmetic here: without it, after three failures Tenacity raises its own RetryError, the calling code catches something other than what it expects, and except requests.exceptions.RequestException does not fire at all.

And the timeout inside the request itself is mandatory. Without it an attempt does not fail — it hangs: there is nothing to retry, because the first attempt has not finished yet.

Java, Resilience4j

The standard choice ever since Hystrix went into maintenance in 2018. The idea is the same, but modular: Retry, CircuitBreaker, Bulkhead and RateLimiter are plugged in separately and combined with each other. It integrates with Spring Boot through annotations, and the retry policy itself lives in application.yml.

@Service
public class OrderService {
    // The annotation says: use the Retry instance named "inventoryService"
    @Retry(name = "inventoryService", fallbackMethod = "fallbackInventory")
    public Inventory getInventory(int productId) {
        return externalInventoryClient.getInventory(productId);
    }
}

The annotation only marks the attachment point. The policy itself — how many times to retry, with what pause, and which errors count as a reason to retry at all — lives in the configuration, and the two are linked by name: name = "inventoryService" in the annotation corresponds to instances.inventoryService in the YAML.

resilience4j:
  retry:
    instances:
      inventoryService:
        max-attempts: 3
        wait-duration: 200ms
        enable-exponential-backoff: true
        exponential-backoff-multiplier: 2
        # the same jitter as in the Python example:
        # without it every client retries in lockstep
        enable-randomized-wait: true
        randomized-wait-factor: 0.5
        retry-exceptions:
          - java.io.IOException
          - java.util.concurrent.TimeoutException
        ignore-exceptions:
          - com.example.InventoryNotFoundException

retry-exceptions and ignore-exceptions are the declarative equivalent of is_retryable from the Python example. The list of exceptions is the policy: leave it unset and Resilience4j will retry everything, business logic errors included. Here InventoryNotFoundException is explicitly ignored — the item does not exist, and it will not appear on the third attempt.

The classes in retry-exceptions have to match what your client actually throws. Feign wraps network failures into FeignException, RestTemplate into ResourceAccessException, and java.io.IOException in the list will never reach them.

If the service still has not answered after all attempts, the fallback is called:

public Inventory fallbackInventory(int productId, Exception e) {
    // ignore-exceptions disables retries, not the fallback:
    // business exceptions land here, and they must be passed through.
    if (e instanceof InventoryNotFoundException notFound) {
        throw notFound;
    }

    log.warn("Inventory service unavailable for product {}: {}",
                productId, e.getMessage());
    inventoryDegraded.increment();   // metric: how often we run degraded

    // Don't invent the stock level. Zero is indistinguishable from a real zero
    // and reaches business logic as the fact "out of stock".
    return Inventory.unknown(productId);
}

ignore-exceptions disables only the retries — the fallback is invoked anyway. That is why business exceptions have to be rethrown by hand, otherwise "item not found" turns into invented data.

Returning zero from a fallback like this is not allowed: the inventory service is merely unavailable, and because of such a stub the shop will show the entire catalogue as sold out.

From the outside none of this is visible. A system running on stubs looks healthy: no errors, and latency is even better than usual — it is not calling the service that is down. So you need a separate metric for the fact of degradation itself, otherwise you will hear about the fake answers from your users.

Retries multiply

Three services in a chain, three attempts each — one user request turns into 27 requests to the bottom service. The very one that is already down. Retries scattered across every layer "just in case" do not add up into protection, they multiply into load: traffic to the failed service grows exactly when it needs to recover.

The cure is that exactly one layer retries. Pick a layer — usually the one closest to the external dependency — and retry only there; the rest propagate the error upward. While you are at it, check that retries are not enabled by default in your HTTP client and in the service mesh: even one retry per layer is already an eightfold amplification.

But even a single layer stays dangerous as long as the number of attempts is fixed. "Three attempts per request" behaves worst precisely under a mass outage: while everything works there are almost no retries, and the moment the service goes down there are three times the normal traffic. A budget breaks that link: count the share of retries in the total number of requests over a window and stop retrying once it crosses a threshold, usually 10–20%. A one-off failure is retried as before, a mass one damps itself.

That is how it is done in gRPC (retryThrottling), in AWS SDK retry quotas and in Envoy's retry_budget. On a bare HTTP client the minimal version is a token bucket: a success deposits a fraction of a token, a retry takes a whole one. That is where "one retry per five successes" at ratio=0.2 comes from. Out of tokens — stop retrying until successes bring them back.

import threading


class RetryBudget:

    SCALE = 1000

    def __init__(self, ratio: float = 0.2, max_tokens: int = 10):
        self._per_success = round(ratio * self.SCALE)
        self._max = max_tokens * self.SCALE
        # start at half: retries work right away
        self._tokens = self._max // 2
        self._lock = threading.Lock()

    def on_success(self) -> None:
        with self._lock:
            self._tokens = min(self._max, self._tokens + self._per_success)

    def try_retry(self) -> bool:
        with self._lock:
            if self._tokens >= self.SCALE:
                self._tokens -= self.SCALE
                return True
            return False       # budget exhausted

The count is kept in whole thousandths of a token rather than in floats. This is not pedantry: at ratio=0.1 ten additions of 0.1 give 0.9999999999999999, and every tenth retry is silently lost. The lock is there for the usual reason: under a thread pool the check and the decrement are not atomic without it, and two threads will spend the same token.

A budget is kept per dependency. A single one for the whole application is meaningless: a failure of one API eats the budget of the others, and retries switch off where everything was fine.

Now it has to be wired into Tenacity. is_retryable alone is no longer enough — the decision depends not only on the error type, but also on the attempt number and on the state of the budget. So instead of retry_if_exception we pass an object of our own: Tenacity accepts any callable in the retry parameter, and hands it the whole state of the attempt.

class BudgetedRetry:

    def __init__(self, budget: RetryBudget, max_attempts: int, name: str):
        self._budget = budget
        self._max_attempts = max_attempts
        self._name = name

    def __call__(self, retry_state) -> bool:
        exc = retry_state.outcome.exception()
        if exc is None or not is_retryable(exc):
            return False

        # Tenacity calls the predicate after the last attempt as well,
        # even though there will be no retry. No reason to spend a token on that.
        if retry_state.attempt_number >= self._max_attempts:
            return False

        if not self._budget.try_retry():
            # This event must be in the metrics: it means the dependency
            # is failing en masse, not once.
            logger.warning("retry budget exhausted for %s", self._name)
            return False

        return True

The final decorator

max_attempts = 3

inventory_budget = RetryBudget(ratio=0.2, max_tokens=10)


@retry(
    stop=stop_after_attempt(max_attempts),
    wait=wait_random_exponential(multiplier=0.1, max=5),
    retry=BudgetedRetry(inventory_budget, max_attempts, "inventory-api"),
    reraise=True,
    before_sleep=before_sleep_log(logger, logging.WARNING),
)
def fetch_data(url):
    response = requests.get(url, timeout=(1, 3))
    response.raise_for_status()
    inventory_budget.on_success()   # a success refills the budget
    return response.json()

on_success() is called after raise_for_status(), not before — otherwise a 500 would count as a success and the budget would be refilled exactly when it must not be.

The resulting behaviour is this: while the service answers, try_retry() almost always returns True — there are plenty of successes and tokens to spare. The moment the service goes down and the successes stop, the accumulated reserve is spent on the first few retries and is never topped up — after that retries switch themselves off. The max_tokens ceiling bounds that reserve: a service that stayed healthy for a day must not earn a day's worth of retries.

Resilience4j has no ready-made budget — its Retry can only count attempts. People build one by hand with the same counter, or push it down to the infrastructure: in Envoy and Istio retry_budget is configured in the sidecar.

Circuit Breaker

Retries help while the outage is short. If a service is down for ten minutes, retries only burn threads: every request honestly waits out its timeout only to get the same error. A circuit breaker removes that wait — after a series of failures it stops calling the troubled service altogether and fails immediately.

That does not free the threads already taken, they hang until their own timeouts. But it stops new ones from being taken, and the caller stays alive while somebody else's problem is being fixed.

There are three states, and the breaker polls nothing on its own — it counts the outcomes of ordinary calls. In Closed calls go through, each outcome is written into a window of recent calls, and when the failure rate in the window crosses the threshold the breaker moves to Open. In Open calls are not executed at all, the exception is raised instantly, and it stays that way for a fixed time — usually tens of seconds. Then Half-Open: a few trial requests are let through, the rest are still rejected. The decision is made on the failure rate among the trials, not on the first failure: with three trials and a 50% threshold one failed call will not open the circuit.

The window matters more than the threshold. Along with the threshold you must always set a minimum number of calls: without it two failed requests at startup make 100% errors and open the circuit out of nowhere. Next comes the window type — count-based or time-based. For sparse traffic a window of 20 calls stretches over tens of seconds and the breaker reacts to things long past; that is where a time-based window belongs.

Network exceptions and timeouts count as failures, obviously; business errors such as validation are better left uncounted, otherwise the circuit opens on a perfectly healthy service. HTTP status codes, however, are a trap of their own: a 500 response is, for the client, a perfectly successful network exchange. Until raise_for_status() is called in Python or recordFailurePredicate is set in Resilience4j, the breaker does not see the 500s and keeps the circuit closed while the service is completely broken.

A failure-rate threshold will not catch a slow service. It answers, but in seconds: no errors, failure rate at zero, circuit closed — while the caller holds threads waiting and goes down itself. So a slow call has to count as a failure too. Resilience4j has a separate pair of parameters for this: a duration threshold and an acceptable share of slow calls. In Python no separate mechanism is needed — with a timeout set, a slow call turns into an exception by itself and lands in the same counter.

Java, Resilience4j

import io.github.resilience4j.circuitbreaker.annotation.CircuitBreaker;
import io.github.resilience4j.circuitbreaker.CallNotPermittedException;

@Service
public class PaymentService {

    // The annotation wraps the method into a circuit breaker named "paymentService"
    @CircuitBreaker(name = "paymentService", fallbackMethod = "paymentFallback")
    public Receipt chargePayment(Order order) {
        // The external call to the payment service
        return paymentClient.charge(order);
    }

    // The circuit is open: the request never left, the money was definitely not taken.
    // Failing here is safe.
    public Receipt paymentFallback(Order order, CallNotPermittedException ex) {
        log.warn("Payment circuit is open, rejecting order {} fast", order.id());
        return Receipt.rejected(order, "payment temporarily unavailable");
    }

    // The call left and never came back. Whether the payment went through is unknown.
    public Receipt paymentFallback(Order order, Throwable ex) {
        log.error("Payment outcome unknown for order {}: {}", order.id(), ex.getMessage());
        return Receipt.pending(order);
    }
}

Resilience4j picks the fallback by exception type: the more specific one wins. There are two of them here because the two cases call for opposite answers.

CallNotPermittedException means the circuit is open and the request never went out to the network: the payment definitely did not go through, and refusing the customer is safe. Any other exception — a timeout, a dropped connection — means the opposite: the request left, there is no answer, and whether the payment went through you do not know.

A single shared fallback would glue the two cases into one answer, and whatever it returned would be wrong half the time. Declare failure — and on a timeout the customer retries and pays twice. Declare success — and the system records a payment that never happened. An open circuit is valuable precisely because it is the one situation where you know for certain that nothing happened.

The policy is set in the config under the same name.

resilience4j:
  circuitbreaker:
    instances:
      paymentService:
        sliding-window-type: COUNT_BASED
        sliding-window-size: 20
        minimum-number-of-calls: 10       # without this the threshold fires on two requests
        failure-rate-threshold: 50
        slow-call-duration-threshold: 2s
        slow-call-rate-threshold: 50      # slow calls count as failures too
        wait-duration-in-open-state: 30s
        permitted-number-of-calls-in-half-open-state: 3
        ignore-exceptions:
          - com.example.PaymentDeclinedException   # "card declined" is not a failure

Python, pybreaker

There are several ready-made implementations — pybreaker and circuitbreaker; the first one is below. The principle is the same: you create a breaker object and decorate the call with it.

import logging

import pybreaker
import requests
from prometheus_client import Counter

logger = logging.getLogger(__name__)

profile_degraded = Counter(
    "profile_degraded_total", "Responses served from a stub instead of user-api"
)


class UserNotFound(Exception):
    pass


breaker = pybreaker.CircuitBreaker(
    fail_max=5,                 # five consecutive failures
    reset_timeout=60,           # after a minute it lets a trial call through
    exclude=[UserNotFound],     # "no such profile" is not a service failure
    name="user-api",
)


@breaker
def fetch_profile(user_id):
    r = requests.get(f"https://api.example.com/users/{user_id}", timeout=(1, 3))
    if r.status_code == 404:
        raise UserNotFound(user_id)
    r.raise_for_status()
    return r.json()


def _stub(user_id):
    profile_degraded.inc()
    return {"user_id": user_id, "name": "Unknown", "stale": True}


def get_user_profile(user_id):
    try:
        return fetch_profile(user_id)
    except UserNotFound:
        # There is no such user - that's an answer, not a failure. Pass it up as is.
        raise
    except pybreaker.CircuitBreakerError:
        # The circuit is open: we never even reached the service
        logger.warning("user-api circuit is open, serving stub")
        return _stub(user_id)
    except requests.exceptions.RequestException as e:
        # The service did not answer; the breaker has already counted this outcome
        logger.warning("user-api call failed (%s), serving stub", e)
        return _stub(user_id)

UserNotFound stands apart. It is passed to exclude, so it does not increment the failure counter: the service is alive and answered on the merits, there is nothing to open the circuit for. And it goes upward as is — "there is no such user" is an answer, not a degradation, and replacing it with a stub would be a lie.

fail_max in pybreaker means consecutive failures, not a failure rate over a window. One success resets the counter: four failures, a success and four more failures will not open the circuit. The model is simpler than Resilience4j's, but also cruder — a service that steadily fails every second request will never open such a breaker. The advice to "open the circuit if 50% of the last 20 requests failed" belongs to Resilience4j; in pybreaker you cannot express it that way.

In Half-Open pybreaker lets a single trial call through: it passes — the circuit closes, it fails — the circuit reopens for another reset_timeout. Here the "first failure decides" model is correct, unlike Resilience4j, where the decision is made on the failure rate among several trial calls.

The breaker lives inside the process

In pybreaker and in Resilience4j alike, the state is kept in process memory by default. If the service runs in fifty pods, the dependency has not one breaker but fifty independent ones: each collects its own statistics and opens on its own.

The threshold is then measured against the traffic of a single instance: a window of twenty calls on a pod that sees one hundredth of the total traffic fills up a hundred times slower, and the breaker reacts late. On the upside, recovery comes out smeared — pods leave Open at different moments, and the load on the recovered dependency ramps up gradually instead of all at once. Either way, you cannot count on the cluster behaving in sync.

Shared state is possible — pybreaker has CircuitRedisStorage — but then Redis sits on the path of every call and becomes a new point of failure. Usually the breaker is left local and the overall picture is watched in the metrics.

That leaves the question of what numbers to put in the config. The threshold should be derived from the normal error level, not from zero: if a service normally serves 2% errors, a 10% threshold is five times above the baseline, and the breaker will stay silent through a real degradation. Look at the error rate distribution over a week and set the threshold above the peak of a normal day, but well below the level at which the dependency stops being useful.

And it belongs on a dependency, not on the service as a whole. One breaker around every call to somebody else's API will open because of trouble on a single endpoint and block the rest. A heavy report and a dozen light reads have different characteristics, which means different breakers.

Fallback and graceful degradation

The first question when something fails is not "what do we return instead" but "can we get the real answer some other way". A standby instance, a second zone, another provider for the same exchange rate — that is not degradation, it is simply a different route to the same data, and if it exists you never get to the fallback.

Degradation starts where there is nowhere to get the real answer. Then the question changes: what do we return instead, so that the user gets at least something and is not deceived in the process.

Four honest answers and one dishonest one

Stale data. A six-hour-old exchange rate, yesterday's catalogue, last hour's search results. Fine when the data changes slowly and the decision based on it is reversible. A staleness limit is mandatory — otherwise one day you will show week-old data — along with a staleness flag that travels upward.

A reduced response. A product page without the "Similar items" block, a feed without personalisation. The best kind of degradation: the user sees less, but everything they see is true.

"Unknown". An explicit "no data" instead of an invented value. It looks worse than a stub, but it is the only answer the calling code can handle correctly.

An explicit failure. "Payment is not possible right now, try again later." The only right answer where there is nothing to invent and nowhere left to degrade.

And the dishonest one: a plausible value. A zero balance, a rate of 1.0, an empty list instead of "we could not compute it". Such an answer is indistinguishable from a real one, passes every check and reaches business logic as a fact. The rule is simple: if you cannot tell from the returned value that it came from a fallback, you must not return it.

A fallback must not mask the problem forever: every time it fires it has to land in a metric, otherwise a system running on stubs looks healthy from the outside. More on that in the monitoring section.

Writing a fallback is easy, it is an alternative branch in a catch. The hard part starts with the question of what exactly to return from it. Often the fallback is attached to a circuit breaker or a retry through fallbackMethod, as in the examples above; without a library you write it by hand.

Python

import logging
import threading
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone

import requests
from prometheus_client import Counter

logger = logging.getLogger(__name__)
rate_degraded = Counter("rate_degraded_total", "Rate served from cache")

API = "https://api.exchangerate.host"
MAX_STALENESS = timedelta(hours=6)


@dataclass(frozen=True)
class Rate:
    value: float
    fresh: bool
    as_of: datetime | None = None


@dataclass(frozen=True)
class Cached:
    value: float
    as_of: datetime

    @property
    def age(self) -> timedelta:
        return datetime.now(timezone.utc) - self.as_of


class RateUnavailable(Exception):
    pass


class LastKnownGood:

    def __init__(self) -> None:
        self._d: dict[str, Cached] = {}
        self._lock = threading.Lock()

    def put(self, key: str, value: float) -> None:
        with self._lock:
            self._d[key] = Cached(value, datetime.now(timezone.utc))

    def get(self, key: str) -> Cached | None:
        with self._lock:
            return self._d.get(key)


cache = LastKnownGood()


def get_exchange_rate(currency: str) -> Rate:
    key = f"rate:{currency}:USD"
    try:
        resp = requests.get(f"{API}/latest?base={currency}", timeout=(1, 2))
        resp.raise_for_status()
        value = resp.json()["rates"]["USD"]
        cache.put(key, value)          # stash it away for the fallback
        return Rate(value=value, fresh=True)

    except requests.exceptions.RequestException as e:
        logger.warning("exchange rate API failed for %s: %s", currency, e)
        rate_degraded.inc()

        cached = cache.get(key)
        if cached is None or cached.age > MAX_STALENESS:
            # Neither fresh data nor acceptably old data. There is no rate - and we won't lie.
            raise RateUnavailable(currency) from e

        # A stale rate is a valid answer, but the caller must know that it is stale
        return Rate(value=cached.value, fresh=False, as_of=cached.as_of)

The cache has a shelf life: six hours is still an exchange rate, thirty is already fiction. If there is no acceptably old data, the function does not return anything plausible — it raises RateUnavailable and lets the caller decide whether to show an error or hide the price block. And the fresh flag travels upward: from a bare number you cannot tell a cached answer from a live one, from a Rate you can.

Java

import org.springframework.web.client.RestClientException;
import java.time.Duration;

@Service
public class RateService {

    private static final Duration MAX_STALENESS = Duration.ofHours(6);

    public Rate getExchangeRate(String currency) {
        try {
            ExchangeRates rates = restClient.get()
                    .uri("/latest?base={cur}", currency)
                    .retrieve()
                    .body(ExchangeRates.class);

            // the body may come back empty - then the next line would throw an NPE
            if (rates == null) {
                throw new RestClientException("empty body");
            }

            double value = rates.rates().get("USD");
            cache.put(currency + ":USD", value);   // stash it away for the fallback
            return Rate.fresh(value);

        } catch (RestClientException e) {
            log.warn("Exchange rate API failed for {}: {}", currency, e.getMessage());
            rateDegraded.increment();

            return cache.get(currency + ":USD")
                    .filter(c -> c.age().compareTo(MAX_STALENESS) < 0)
                    .map(c -> Rate.stale(c.value(), c.asOf()))
                    // neither fresh data nor acceptably old data - there is no rate
                    .orElseThrow(() -> new RateUnavailableException(currency, e));
        }
    }
}

The RestClient timeout is set on the client rather than in the call chain, and applies to every request made through it at once:

import org.springframework.http.client.SimpleClientHttpRequestFactory;
import org.springframework.web.client.RestClient;

import java.time.Duration;

@Configuration
public class RateClientConfig {

    @Bean
    public RestClient rateRestClient() {
        SimpleClientHttpRequestFactory requestFactory = new SimpleClientHttpRequestFactory();
        requestFactory.setConnectTimeout(Duration.ofSeconds(1));
        requestFactory.setReadTimeout(Duration.ofSeconds(2));

        return RestClient.builder()
                .baseUrl("https://api.exchangerate.host")
                .requestFactory(requestFactory)
                .build();
    }
}

Far from everything can degrade. Analytics, recommendations, reviews, view counters — stale data offends nobody there. But the outcome of a payment is never replaced by a guess under any circumstances: there it is either the real answer or an honest "we don't know", and never a plausible number.

The fallback is the least tested code in the system

It only runs on failure, which means it is almost never exercised in tests, and it fires exactly when everything is on fire.

A fallback must not depend on anything that can go down together with the main path. A cache in Redis is a network dependency too — if Redis died along with the API, your except will raise an exception from inside the exception handler. A local copy of the last known value in process memory is more reliable than an external cache precisely because it has nothing that can fall over.

And the cold start. Right after a deploy the cache is empty, and the branch that fires is not "serve the old value" but "there is no data at all".

Checking this is cheap: toxiproxy in front of the dependency and three scenarios — the dependency is down, the dependency and the cache are down together, the cache is empty.

When there are many degradations at once

One degradation goes unnoticed; three at the same time are a different page altogether. If recommendations, reviews and stock levels all fall away at once, the user gets a 200 OK and an almost empty screen. Formally the service works, in fact it does not.

So the share of fallback answers is worth counting not per dependency but per request in total — and setting a threshold beyond which it is more honest to return an error than to pretend you are working.

Telling that answer apart from a real one is the most awkward of the questions you can ask about a fallback, and the most useful.

Timeouts and deadlines

The service from the first paragraph — the one that started answering in three seconds instead of thirty milliseconds — is dangerous for exactly this reason. A call without a timeout does not fail, it hangs, holding a thread or a connection from the pool. An error is visible immediately; a hung call is invisible until you run out of threads.

An HTTP call has two timeouts, and they are about different things. The connect timeout is short, 0.5–1 second: within that time either there is a TCP connection or the host is not there, and waiting longer buys nothing. The read timeout is longer, 2–5 seconds depending on circumstances — that is the time for the actual work. In the examples above these are timeout=(1, 2) in Python and the setConnectTimeout / setReadTimeout pair in Java.

With a database it gets trickier: there are three timeouts, and mixing them up is expensive. connectionTimeout in HikariCP is how long to wait for a free connection from the pool; it fires when the pool is exhausted and has nothing to do with how long the query takes. Statement.setQueryTimeout or statement_timeout on the Postgres side is how long the query itself runs. socketTimeout in the driver is how long to wait for a network answer if the server died silently. All three have to be set: without the first a thread hangs in the queue for a connection, without the second a heavy query holds a connection to the bitter end, without the third nobody notices a broken connection.

But the most common mistake is not in the numbers — it is that timeouts add up. A waits for B for three seconds, B waits for C for ten, and B keeps working long after A has left and its answer is of no use to anyone. Picking better numbers does not fix this: a chain longer than two links will always find a way to get out of sync.

What has to be passed down is not the timeout but the remaining time. The top level fixes a deadline for the whole operation and tells every call how much time is left. Service B, having received "1.2 s remaining", will not wait ten seconds for C — it sets a timeout no larger than what is left, and if nothing is left it fails immediately without wasting the call.

In gRPC this is built in: the deadline travels in the metadata and propagates along the chain by itself. Over plain HTTP you have to do it by hand:

Python, asyncio + aiohttp

import asyncio

import aiohttp

DEFAULT_BUDGET = 3.0


def budget_seconds(request) -> float:
    header = request.headers.get("X-Request-Deadline-Ms")
    if header is None:
        return DEFAULT_BUDGET
    return min(DEFAULT_BUDGET, int(header) / 1000)


async def get_json(session, url: str, budget: asyncio.Timeout) -> dict:
    left = budget.when() - asyncio.get_running_loop().time()
    headers = {"X-Request-Deadline-Ms": str(int(left * 1000))}
    async with session.get(url, headers=headers) as resp:
        resp.raise_for_status()
        return await resp.json()


async def handle(request, session):
    async with asyncio.timeout(budget_seconds(request)) as budget:
        user = await get_json(session, USERS, budget)
        # parallel calls share the same budget
        cart, orders = await asyncio.gather(
            get_json(session, CART, budget),
            get_json(session, ORDERS, budget),
        )
    return {"user": user, "cart": cart, "orders": orders}

There is no need to compute what is left and compare timeouts by hand — asyncio.timeout does it for you. Nested blocks combine correctly: if the outer budget is 3 seconds and an inner call brings its own timeout of 10, the outer 3 wins — asyncio tears down the whole task tree, including the ones started through gather. A separate ClientTimeout for aiohttp is not needed either: the cancellation reaches the socket.

Exactly one thing is done by hand, the one thing asyncio cannot know about — passing the deadline over the network. Outward, the remaining time goes out as a header, and budget.when() gives the absolute end time. Inward, budget_seconds() reads somebody else's deadline and takes the minimum with its own: if the caller is only willing to wait a second, waiting three is pointless.

Java, OkHttp

import java.io.InterruptedIOException;
import java.time.Duration;

OkHttpClient client = new OkHttpClient.Builder()
        .connectTimeout(Duration.ofSeconds(1))
        .readTimeout(Duration.ofSeconds(3))
        .writeTimeout(Duration.ofSeconds(3))
        // total budget for the whole call, including DNS, redirects and OkHttp's own retries
        .callTimeout(Duration.ofSeconds(5))
        .build();

Request request = new Request.Builder()
        .url("https://api.github.com/repos/user/repo")
        .build();

try (Response response = client.newCall(request).execute()) {
    if (!response.isSuccessful()) {
        // handle the error code
    }
    String body = response.body().string();
} catch (InterruptedIOException e) {
    // both SocketTimeoutException and a callTimeout firing land here
    log.warn("GitHub API did not respond in time: {}", e.getMessage());
}

callTimeout matters more than the other three: without it connect, read and write add up, and the real wait turns out longer than the calling code assumes. What you have to catch is InterruptedIOException, not SocketTimeoutException — a callTimeout firing throws exactly that, and a narrow catch will miss it.

There is no equivalent of asyncio.timeout in Java: the deadline has to be tracked yourself and applied to every call separately. The closest thing in spirit is passing the remaining budget as a parameter and subtracting what has been spent, just like in the Python example, only manually.

A timeout stops the waiting, not the work

When a client gives up on a timeout, the server does not learn about it. It keeps executing the request, writing to the database, taking the money — the answer just goes nowhere. A timeout is a decision to stop waiting, not a cancellation of the operation.

Hence the consequence that breaks the most systems: a timeout does not mean the operation did not happen. It means you do not know its outcome. The same payment case as in the breaker section: after a timeout you may only retry idempotent operations — more on that next.

And the converse: if the client is gone, there is usually no point in working on. That is exactly what the deadline is passed down the chain for — so that the service below can check what is left and not start something nobody will wait for.

A timeout is set slightly above the ninety-ninth percentile of a normal day, usually two or three p99s. For a service with a typical 100 ms and a p99 of 300 ms a reasonable value is about a second. Five seconds at the same 100 ms is not headroom, it is a decision to hold a thread five times longer than it takes to conclude that no answer is coming. The average response time is no use here at all — it is the tail of the distribution that decides how many threads you lose during a degradation.

And check the multiplication. A timeout is multiplied by the number of attempts: three retries of three seconds is nine seconds in the worst case, not three. If there is an operation deadline on top, it has to account for the retries, otherwise it is not a deadline but a wish.

Idempotency: operations without a repeated effect

The previous section ended on the fact that after a timeout the outcome of an operation is unknown. There is exactly one case in which retrying it is safe: when the retry adds nothing to the first call.

PUT /users/42 {"name": "Ivan"} is idempotent — repeat it as many times as you like, the name becomes "Ivan" and stays that way. POST /orders is not: every call creates a new order. GET and DELETE are idempotent by the standard as well, but POST is the interesting one, because taking money and sending email live there.

There is one way to make POST idempotent: the client sends an operation identifier and the server remembers the result under it. An Idempotency-Key header, a message ID from a queue, a saga ID — all that matters is that it comes from outside and does not change on a retry. The first request creates the resource, the second one with the same key gets the already created one.

The most reliable place to store this is a unique index, in the same transaction as the effect itself. Then it is the database that does the rejecting: ON CONFLICT DO NOTHING, or a caught uniqueness violation, reads as "already done". Checking "is there such a key already?" with a separate query before the insert does not work, for the same reason as everything else in this section.

Python

from dataclasses import dataclass

import psycopg

INSERT_IF_ABSENT = """
    INSERT INTO users (username, email)
    VALUES (%(username)s, %(email)s)
    ON CONFLICT (username) DO NOTHING
    RETURNING id, username, email
"""

FIND_BY_USERNAME = """
    SELECT id, username, email FROM users WHERE username = %(username)s
"""


@dataclass(frozen=True)
class User:
    id: int
    username: str
    email: str


def insert_if_absent(conn: psycopg.Connection, username: str, email: str) -> User | None:
    with conn.cursor() as cur:
        cur.execute(INSERT_IF_ABSENT, {"username": username, "email": email})
        row = cur.fetchone()
        return User(*row) if row else None


def find_by_username(conn: psycopg.Connection, username: str) -> User | None:
    with conn.cursor() as cur:
        cur.execute(FIND_BY_USERNAME, {"username": username})
        row = cur.fetchone()
        return User(*row) if row else None


def create_user(conn: psycopg.Connection, username: str, email: str) -> User:
    created = insert_if_absent(conn, username, email)
    if created is not None:
        return created

    # The insert did not go through - so the user is already there
    existing = find_by_username(conn, username)
    if existing is None:
        # The row was deleted between the two queries - rare, but possible
        raise RuntimeError(f"user {username!r} vanished between insert and read")
    return existing

The work here is done not by the code but by the word UNIQUE in the schema. A separate SELECT and INSERT do not survive the race: a second request fits between them, both see nothing and both insert. On a live Postgres with sixteen concurrent calls this produces duplicates more than half the time. ON CONFLICT DO NOTHING does both in a single statement, and the decision is made by the database on its own index, where there is no room to squeeze in — only, when DO NOTHING fires, RETURNING has nothing to return, so the existing row is read with a second query.

And a caveat: the function is idempotent by username, not by the pair with email — a repeat call with a different address returns the old user without updating them. That is a line in the API contract, not an implementation detail.

Between "not started" and "done" there is a third state

An operation has three states, not two: not started, in progress, finished. A "done already?" flag cannot tell the third one apart, and it breaks exactly where idempotency is needed most.

What do you return if a repeat request with the same key arrives while the first one is still running? Not "already done" — that is untrue. And not executing it a second time either. What is left is "the operation is in progress, ask later": 409 Conflict for a synchronous API, a stored status for an asynchronous one.

So what is stored under the key is a status, not a flag, and it changes according to the outcome of the call, not before it.

Java

import org.springframework.dao.DuplicateKeyException;

@Service
public class PaymentProcessor {

    public PaymentResult process(String txId, Order order) {
        // 1. Claim the txId. A unique index in the database is the only thing that
        //    really protects against the race: a local Set does not survive a restart
        //    and is invisible to the second instance of the service.
        try {
            attempts.insertStarted(txId, order.id());
        } catch (DuplicateKeyException duplicate) {
            // This payment has already been started. Return its outcome without redoing it.
            return attempts.resultOf(txId);
        }

        // 2. The external call - with the same txId as the idempotency key at the provider
        try {
            GatewayReceipt receipt = gateway.charge(txId, order.amount());
            attempts.markSuccess(txId, receipt.id());
            return PaymentResult.success(receipt.id());

        } catch (GatewayDeclined declined) {
            // The provider said "no" - that is a final answer
            attempts.markFailed(txId, declined.reason());
            return PaymentResult.declined(declined.reason());

        } catch (GatewayUnavailable unknown) {
            // There is no answer. Whether the payment went through we do not know, and we
            // have no right to either confirm or cancel it. The row stays STARTED,
            // a background reconciliation with the provider will resolve it.
            attempts.markUnknown(txId, unknown.getMessage());
            return PaymentResult.inProgress();
        }
    }
}

Rows stuck in STARTED are not a leak, they are a way of storing "I don't know". The reconciliation that resolves them is something you will have to write, otherwise they pile up forever. In Python it is the same table with a status column, only with ON CONFLICT DO NOTHING instead of catching DuplicateKeyException.

Bulkhead (limiting concurrent calls, semaphores)

A timeout limits one call, not the number of them. The service from the first paragraph answered in three seconds instead of thirty milliseconds — with a five-second timeout every call does finish honestly, except that while it lasts the thread is busy. A hundred concurrent requests to a dependency like that means a hundred busy threads, and there is nobody left to serve anything else.

A bulkhead puts a bound on the number of concurrent calls rather than on their duration. The reporting service, the one that can hang for thirty seconds, is given five threads out of fifty — and it will take only those, no matter how many requests arrive. The remaining forty-five keep serving whatever still works.

In practice this is either a separate thread pool for a particular dependency, or a semaphore if the code is asynchronous. A pool also moves the call off the main thread; a semaphore simply does not let more than N tasks run at once.

Against a breaker the roles differ: the breaker decides whether to call the dependency at all, the bulkhead decides how many requests to let through at once. A breaker is useless while the dependency answers but slowly: there are no errors, the circuit is closed, and the threads run out. That is exactly the case a bulkhead covers.

The limit is computed, not guessed: requests per second times the response time. 50 rps at 200 ms means ten requests inside the dependency at any moment, and that is the working limit. Setting a hundred is pointless: the extra ninety will queue up on the other side anyway.

Python, asyncio + Semaphore

import asyncio
from contextlib import asynccontextmanager

import aiohttp

REPORTS = asyncio.Semaphore(10)
MAX_WAIT = 0.5


class ServiceBusy(Exception):
    pass


@asynccontextmanager
async def bulkhead(sem: asyncio.Semaphore, max_wait: float, name: str):
    try:
        async with asyncio.timeout(max_wait):
            await sem.acquire()
    except TimeoutError:
        raise ServiceBusy(name) from None
    try:
        yield
    finally:
        sem.release()


async def fetch_report(session: aiohttp.ClientSession, url: str) -> dict:
    async with bulkhead(REPORTS, MAX_WAIT, "reports"):
        async with session.get(url) as resp:
            resp.raise_for_status()
            return await resp.json()

The key line here is not the semaphore itself but the timeout around acquiring it. Waiting in the queue is no different from waiting for the answer: the request still stands there, still holds memory, still creeps toward the client's timeout. A semaphore without a bound on the wait gives you exactly half a bulkhead: concurrency stays under the limit, and the number of waiters is bounded by nothing at all.

Java

import java.time.Duration;
import java.util.concurrent.*;

// A separate pool for slow reports: 5 threads, a queue of 10.
// The queue is bounded on purpose: an unbounded one is a deferred OOM.
static final ExecutorService REPORTS = new ThreadPoolExecutor(
        5, 5, 0L, TimeUnit.MILLISECONDS,
        new ArrayBlockingQueue<>(10));

static CompletableFuture<Object> reportAsync(String id) {
    try {
        return CompletableFuture.supplyAsync(() -> fetchReport(id), REPORTS);
    } catch (RejectedExecutionException busy) {
        // The bulkhead is full. Under load this is a normal answer, not an outage.
        throw new ServiceBusyException(Duration.ofSeconds(1));
    }
}

A pool of 5 threads with a queue of 10 accepts exactly 15 tasks; the sixteenth gets a RejectedExecutionException — synchronously, out of supplyAsync, not deferred inside the CompletableFuture. On fifty concurrent requests that comes out as 15 accepted and 35 rejected in 7 ms. It is the boundedness of the queue that makes a bulkhead a bulkhead: without it excess work is not shed, it accumulates.

In Resilience4j the same effect is available through the Bulkhead module. For example:

import io.github.resilience4j.bulkhead.*;
import java.time.Duration;
import java.util.function.Supplier;

BulkheadConfig config = BulkheadConfig.custom()
        .maxConcurrentCalls(10)          // the default is 25, not 5
        .maxWaitDuration(Duration.ZERO)  // no queue: reject straight away
        .build();

Bulkhead bulkhead = Bulkhead.of("reportService", config);

Supplier<Object> decorated =
        Bulkhead.decorateSupplier(bulkhead, () -> fetchReport("42"));

You most likely have one bulkhead already: the database connection pool is the very same semaphore — maximumPoolSize in HikariCP, limit_per_host in aiohttp. The trouble is not the absence of a limit but the fact that there is only one: a heavy report and an order checkout take connections from the shared pool, and the first one eats it whole. A separate small pool for heavy queries is the cheapest bulkhead you can put in today.

And check the sum: four bulkheads of 50 each with a hundred threads available isolate nothing — any two of them will exhaust the resource completely.

A bulkhead rejection is a normal answer, not an outage. BulkheadFullException and RejectedExecutionException mean "no room right now", not "the dependency is broken": outward that is a 503 with Retry-After, in the logs a WARN, in the metrics a counter of its own. Growing steadily — the limit is too low or the dependency has degraded. Never growing — the bulkhead is not limiting anything, and it is worth checking with a load test.

It is also worth looking at how a bulkhead rejection is accounted for by the breaker. By default Resilience4j treats any exception as a failure, BulkheadFullException included. Limit of 10, forty concurrent requests: ten went through, thirty were turned away by the bulkhead — a 75% failure rate, and the breaker opened. The next wave does not reach the dependency at all, even though it was healthy the whole time: it was switched off by our own concurrency limit. The order of the wrappers is right, the bulkhead does belong inside the breaker — what needs fixing is the accounting:

import io.github.resilience4j.bulkhead.BulkheadFullException;
import io.github.resilience4j.circuitbreaker.CircuitBreakerConfig;

CircuitBreakerConfig config = CircuitBreakerConfig.custom()
        // a bulkhead rejection is not a dependency failure, the breaker must not count it
        .ignoreExceptions(BulkheadFullException.class)
        .build();

Now the breaker stays closed and the next wave is served as usual.

Retrying a request that a bulkhead turned away is almost always pointless too: a rejection means "there is no capacity", not "bad luck", and capacity will not appear within 50 ms — the retry will only eat into the retry budget.

Monitoring, logging and alerts

Every pattern from the previous sections makes failure less visible. A retry hides a one-off error, a breaker hides a dependency that is down, a fallback hides missing data. That is their job, and precisely why the usual metrics stop telling the truth: the error rate drops, latency improves, uptime is green, and half of the answers are assembled from a cache.

What you end up having to measure is what ordinary monitoring does not have: the share of retries in the total number of requests and the cases where the budget ran out; breaker transitions between states and the time spent in Open; the share of answers served from a fallback — per dependency and per request in total; bulkhead rejections and queue length. None of these appears on its own: the counter is placed by hand, in the same code where the fallback is written.

Alert on a share, not on an event. One timeout at night is weather, not an incident — retries exist exactly so that this does not wake the on-call engineer. An incident is when the share of degraded answers stays above the threshold for several minutes in a row, or when a breaker fails to close for longer than the usual recovery time.

It is also worth checking that the alert fires at all. Toxiproxy in front of the dependency, the same scenarios as for the fallback, and watch not only how the code behaves but whether anything lights up on the dashboard. A metric never exercised under failure is no better than no metric.

And monitoring itself must not become a dependency. Sending logs synchronously to a remote server inside request handling means a stalled collector will slow the application down: observability takes down the thing it observes. Keep the logger asynchronous and the buffer local — losing some logs on overflow beats blocking.

Conclusion

None of these patterns removes failures. They change the shape of failure: instead of a hung request, a fast error; instead of a cascade, degradation of a single area; instead of uncertainty, an explicit "unknown" in the answer. Failure does not go away — it becomes something you can plan for in code.

The price is new decisions, every one of which can be made wrong. Retries without a budget multiply the load onto exactly the service that is down. A fallback returning zero instead of "I don't know" lies to your business logic more convincingly than an exception ever could. A breaker with a threshold measured from zero opens on a healthy service, and one with a threshold picked at random stays silent through a real degradation. There is not a single setting here whose default will fit your system.

Which leaves one thing worth checking as soon as everything is written: is the degradation visible in the metrics. No errors, excellent latency, green graphs — and that is exactly what a system answering with stubs looks like. The better your fallbacks are written, the later you will notice it, and the more likely it is that whoever notices will not be you.

Distributed Transactions in Microservices: From SAGA to Two-Phase Commit

Timofei Ivankov — Mon, 10 Aug 2026 13:39:52 +0000

The transition from a monolith to a microservice architecture brings flexibility and scalability, but also creates new challenges. One of the key issues is data consistency and transactions. In a monolith, you can typically wrap multiple operations in a single ACID transaction: either all operations succeed, or an error triggers a full rollback. In the world of microservices, this straightforward approach doesn't work. Each service is autonomous, each has its own database, and they communicate over a network. As a result, guaranteeing the atomicity and integrity of processes spanning multiple services is difficult. This creates the risk of partial updates: one part of the system changes while another doesn't, causing data to drift apart.

To address this, several patterns and protocols have been developed. Below are two poles — saga and two-phase commit — and what lies between them: TCC and Outbox. Separately, there's isolation: the single letter of ACID that a saga completely loses. This is usually ignored in articles about saga, and the cost is paid in production.

Limitations of ACID transactions in microservice architecture

ACID is the classic set of properties of a local transaction: Atomicity, Consistency, Isolation, and Durability. In the context of a single database, ACID guarantees that a transaction is indivisible — either executed entirely or rolled back without a trace, transitioning the database from one consistent state to another. However, in a microservice architecture, where data is distributed across multiple services and databases, ensuring an ACID model "system-wide" is extremely difficult.

The data is physically separated: each service has its own database, and a transaction that affects two of them cannot be local. Built-in DBMS mechanisms do not work across network boundaries. The interaction itself occurs over the network — REST, gRPC, messages — with all the expected delays and interruptions. So, attempting to chain operations together as a single transaction runs into the problem that some operations are committed and others are not. There's also no isolation between services: without a shared transaction manager, one service can read another's intermediate state.

This is all well-known. But the following is usually formulated incorrectly.

The cost of coordination, not just the CAP theorem. CAP describes a narrow scenario — what to do during a network partition: maintain consistency and deny service, or respond, risking data divergence. A global ACID transaction chooses consistency here. But it incurs the main cost in normal operation, and CAP is silent about this. PACELC states it more completely: if Partition then A or C, else L or C — during a partition, we choose between availability and consistency, and the rest of the time, between latency and consistency. For 2PC, the latter half is more important than the former. Two network round trips per transaction are latency that every operation pays, even when the network is perfect. Plus, availability arithmetic: if each of the five participants is available 99% of the time, a transaction requiring agreement from all five is available 0.99⁵ ≈ 95% of the time — degradation without a single connection interruption. A bottleneck in a high-load system occurs not during a crash, but constantly.

Classic ACID transactions "across service boundaries" are either impossible without special protocols or lead to serious scalability and fault tolerance issues. A distributed system needs different approaches to consistency. Next, we'll consider two main solutions: the Saga pattern, based on transaction splitting and compensation, and the 2PC protocol, which coordinates atomic confirmation.

Saga Pattern: Distributed Transactions via Compensating Actions

A saga breaks a large business transaction into a sequence of local ones: each step is committed to its own service and immediately becomes visible. There is no global transaction at all, so there is no global rollback. If all steps are successful, the whole saga is considered successful. If an error occurs at some stage, the Saga pattern runs compensating transactions to undo actions already completed and return the system to a consistent state.

Motivation and how Saga works

A saga solves the problem of a business operation requiring data changes in multiple services. Example: placing an order in an online store — you need to create an order in the Order service, charge the payment in the Payment service, and reserve the product in the Inventory warehouse. In a monolith, we would do this in a single transaction. In microservices, a saga allows you to achieve a similar effect by sequentially executing local transactions.

The steps occur in a predetermined order, and each is a regular local transaction in its own service. Order creates an order and publishes OrderCreated, Payment debits the funds on that event, and Inventory reserves the product. While everything is running, the saga simply moves forward.

It breaks on the first failure. If Inventory responds that the product is out of stock, there's nothing to roll back: both previous transactions have long been committed. Instead of a rollback, compensation is triggered: Payment returns the money, and Order marks the order as "canceled." Compensation is the same transaction as everything else; it just does the opposite.

This result is often called "saga atomicity." More precisely, a saga provides ACD without I: atomicity is simulated by compensation, consistency and durability are provided by local transactions, and there is no isolation at all. There is no "without a trace" rollback — compensation is a new transaction that performs the reverse action, and the intermediate state becomes visible to others in the meantime. A canceled order remains in history with the "canceled" status, rather than disappearing.

Three types of steps: compensatable, pivot, retriable

Not all saga steps are the same, and before writing compensations, they should be sorted into three categories.

Compensatable — a step that can be undone by a reverse action. Order creation is canceled by changing the status, a reserve in the warehouse is removed, and a card hold can be released as long as the funds haven't been captured yet: the money hasn't been sent anywhere; the bank simply releases the frozen amount. Compensating transactions are written only for steps in this category.

Pivot — the point of no return. This is either the last compensatable step or the first non-compensatable step. Once it's committed, the saga cannot be rolled back: the only option is to move forward.

Retriable — everything that comes after the pivot. Such steps must eventually succeed, and they are never undone, only repeated. Examples include sending an email, publishing an event, or accruing bonuses.

Hence the rule by which any saga is designed: the order of steps must be compensatable → pivot → retriable, and nothing else. If a non-compensatable step is in the middle, the saga is broken at the design level; if the next step fails, there will be nothing to roll it back with.

How to find a pivot in your saga: go through the steps and at each one, ask — if this step has been completed, are we ready to cancel it automatically, without human intervention? The first "no" is the point of no return. The answer is provided by the business, not the code: if the payment provider has idempotent reversals and the business allows them without confirmation, the charge remains compensatable; if the reversal is subject to manual review, the charge becomes a pivot, and everything after it must be retriable.

A practical trick for when a pivot occurs too early: split the step into two phases. This is precisely why payments are almost always divided into authorization and capture — authorization is compensated by lifting the hold, while capture can be postponed until the very end. The later the point of no return, the more room the saga has for rollback.

And the opposite requirement applies to retriable steps: they cannot refuse for business reasons. If a step after a pivot can respond "no," it is not retriable, and the saga is designed incorrectly. Network errors and service unavailability can be retried, but domain-rule rejections cannot.

The most common mistake here seems harmless: a notification is sent as a regular saga step, and its failure triggers a rollback. Formally, everything is logical: the step failed, so we roll it back. In essence, we refund the client and cancel the order because the email was not sent.

Orchestration vs. choreography

The logic of a sequence must live somewhere, and there are exactly two options.

With orchestration, it is handled by a separate component — the saga orchestrator. It knows the entire scenario, calls Service A, then Service B, and triggers compensation itself if an error occurs. The entire sequence is read in a single file, but another service is added to the system that knows about all the others.

With choreography, the scenario doesn't exist anywhere; there are only reactions. The Order Service publishes OrderCreated, the Payment Service picks up and responds with PaymentApproved or PaymentFailed, the Order Service listens and decides whether to confirm or cancel the order. There's no central component, but to understand what actually happens when an order is placed, you'd have to open four repositories.

Choreography is often sufficient in simple systems, but orchestration can be more convenient for complex processes with multiple conditions.

What it costs

A saga has one advantage that an ACID transaction fundamentally can't have: compensation doesn't have to be an exact reversal. If a payment can't be automatically reversed, the compensation is marking the order "requires manual review." ACID rollback can't do this; it either cancels everything or nothing. The other advantages have already been mentioned above: there are no global locks, no coordinator, and services don't wait for each other.

Now the bill, and it's quite long.

Compensation must be written for each compensatable step, and this isn't a mechanical task: you have to decide what the system does if step X succeeds but step Y fails. "Cancel a payment" is a business decision, not a line of code, especially once the funds have already been debited.

Everything, including compensation, must be idempotent. Retries and duplicate messages are inevitable in a distributed environment: the service will receive the same event twice, and compensation will be retried after a failure. The "already done?" check against internal status is not enough — two retries arrive simultaneously, and both see "not done." Only an external operation identifier in a unique index works.

Debugging is more difficult than in a monolith: the sequence is not visible anywhere, especially with choreography. A correlation identifier in all messages and distributed tracing are essential from day one — without them, the saga simply cannot be reconstructed after an incident.

The three remaining cost items are discussed separately below: intermediate inconsistency in the next section, progress storage in the code example, and unsuitability for operations requiring instant atomicity in the section on 2PC.

No isolation: what this means in practice

A saga loses exactly one of its four ACID letters, and that's no small thing. The lack of isolation means that the saga's intermediate results are visible to everyone else: another saga, a background job, or a regular query reads the data when half the steps are completed and half are not. This is impossible in a local transaction; the DBMS is responsible for that. In a saga, responsibility shifts to you, and if you don't explicitly take it, the system suffers from three classic anomalies.

Lost updates. A saga overwrites a change someone made while it was running. A user placed an order, the saga reached the payment step, at which point the user clicked "cancel," the order was marked as canceled, and the next saga step overwrote it with "confirmed." The cancellation disappeared, and no error occurred: every local transaction executed correctly.

Dirty read. Someone makes a decision based on data that the saga will later reverse. An order is created, funds are debited, the loyalty service sees the debit and credits cashback. Two seconds later, the saga crashes when reserving the item, the funds are returned, and the cashback remains. The compensation canceled the debit, but didn't cancel someone else's decision based on it.

Unrepeatable read. Two steps in the same saga read the same thing and see different results. The first step checked that there are sufficient funds in the account, the third debits it, and in between, another transaction occurred, and the funds are gone. The check performed at the beginning of the saga is out of date by the time the action is taken.

All three have one thing in common: none of them show up in happy path tests, and none of them produce errors in the logs.

Countermeasures

The set of techniques here has long been established, and you have to choose deliberately; there is no universal solution.

Semantic locking. The record is marked with an "in progress" flag — PENDING, PROCESSING, RESERVED status. Anyone reading it must respect this flag: wait, refuse the user, or show the data with a caveat. This is the most common countermeasure, but it has a price that is often overlooked: you need to decide what the reader does when they see the flag, and a timeout is essential, otherwise a failed saga will leave the record locked forever.

Commutative updates. Operations for which order is irrelevant: balance = balance - 100 instead of balance = 900. Then a lost update is impossible in principle, and compensation becomes trivial — add back. This is the cheapest technique of all: it requires nothing more than writing deltas instead of absolute values.

Pessimistic step ordering. Sometimes an anomaly is easier to fix by reordering the steps than by protecting against it. If cashback is awarded after the entire saga, rather than after the debit, the dirty read in the example above disappears on its own. Reordering doesn't provide formal guarantees, but it significantly reduces business risk and doesn't require a single line of infrastructure code.

Re-read before write. Before writing, the step rereads the record and checks that it hasn't changed since it read it. If it has, the saga is aborted and restarted. This is essentially standard optimistic version locking, and it works against lost updates.

Version log. Operations are written to a log and applied in order, even if they arrive out of order. If an order cancellation arrives before its creation, both operations will be logged and applied correctly. This technique converts non-commutative operations into commutative ones, at the cost of additional storage.

Choose the mechanism by what's at stake. A system-level, not a step-level strategy: the mechanism is chosen based on the cost of error. A thousand-ruble transfer is processed by a saga, while a million-ruble transfer goes through a strict transaction or manual confirmation. This is an admission that a single consistency model for all operations is a compromise, and somewhere it will be the wrong one.

Where to start

First, it's worth checking whether operations can be made commutative; it costs nothing and eliminates a whole class of problems. Place semantic locks on the records the saga holds between steps and immediately set a timeout for them. Next, reread the value where a concurrent write is likely. And check the order of the steps separately: some anomalies can be resolved with a simple rearrangement.

The main thing is not to treat this as an additional reliability feature that can be added later. The lack of isolation doesn't manifest itself under low load and doesn't break tests. It manifests itself in production as discrepancies that are impossible to reproduce.

Example: a saga for placing an order

Let's take the same scenario: create an order, take payment, send confirmation. A full-fledged saga engine won't fit in this article, but it doesn't need to be shown in its entirety — two things are enough: how the saga is described and what the engine does with this description.

A saga is defined by a list of steps. Each step has an action, a compensation, and a type — the same compensatable / pivot / retriable discussed above. The type determines what happens in the event of a failure.

// Step = action + compensation + type.
// The type decides what the engine does on failure.
List<SagaStep> ORDER_SAGA = List.of(

    step("create-order", COMPENSATABLE,
         ctx -> ctx.put("orderId", orders.create(ctx.sagaId(), ctx.payload())),
         ctx -> orders.cancel(ctx.sagaId(), ctx.getLong("orderId"))),

    step("authorize-payment", COMPENSATABLE,
         ctx -> ctx.put("authId", payments.authorize(ctx.sagaId(), ctx.amount())),
         ctx -> payments.releaseHold(ctx.sagaId(), ctx.get("authId"))),

    // the money is gone - no rollback from here on
    step("capture-payment", PIVOT,
         ctx -> payments.capture(ctx.sagaId(), ctx.get("authId")),
         ctx -> { throw new UnsupportedOperationException("a pivot cannot be compensated"); }),

    // the email failed - not a reason to refund
    step("send-confirmation", RETRIABLE,
         ctx -> notifications.confirm(ctx.sagaId(), ctx.getLong("orderId")),
         ctx -> { throw new UnsupportedOperationException("retriable steps are only retried"); })
);

Pay attention to the payment. The authorization is compensatable — the bank simply removes the hold; the money never went anywhere. The capture can't be compensated for, so it's declared the point of no return, and everything after that must be retriable. Because of this, a failure to send the email will result in a new attempt to send it, rather than a cancellation of the paid order.

Now the engine. The saga progress is stored in a standard table: saga ID, current step number, accumulated context, next attempt time. A background worker selects the sagas that are due and advances each one exactly one step:

// One tick: take a saga, run the next step, persist the progress
void advance(SagaRecord saga) {
    SagaStep step = steps(saga).get(saga.step());
    try {
        // network call, Idempotency-Key = sagaId
        step.execute(ctx);
        // local commit: step number + context
        repo.save(saga.advanced(ctx));

        // If the process dies between these two lines, the step runs again.
        // That is not a bug, it is the contract: execute must be idempotent.

    } catch (TransientFailure e) {
        // timeout, 5xx, dropped connection - retry later
        repo.save(saga.retryLater(backoff(saga.attempts()), e));
    } catch (BusinessRejection e) {
        // a business "no" - compensate, but no further back than the pivot
        repo.save(saga.startCompensation(e));
    }
}

Ten lines, and they contain almost everything that distinguishes a working saga from a textbook one. The state survives process restarts — a crashed worker will recover and continue from the same step. Failures are divided into two kinds, and confusing them is expensive: a timeout doesn't mean the operation failed. A rollback on timeout compensates a payment that actually went through, and the money is returned to the client twice. And an idempotency key is required not for aesthetics, but because a retry is guaranteed to happen.

A lot remains behind the scenes: the reverse pass over compensations, escalation to manual review when compensation fails after N attempts, deduplication on the receiving end, and a deadline for the saga as a whole. But it is this scaffolding that makes up the bulk of the code, not the business logic. Ready-made solutions handle exactly this: Camunda, Temporal and Cadence workflow engines, long-running actions (LRA in MicroProfile), Seata in SAGA mode. Inside they are all the same: a sequence of steps, compensations, and durably saved progress. The only question is whether you write it yourself or get it ready-made.

Two-Phase Commit (2PC)

Saga tolerates the intermediate state being visible. 2PC doesn't: it coordinates multiple nodes through a central coordinator so that only the final result is visible — either everyone commits or everyone rolls back.

How 2PC works

There can be two or more participants — services, databases, anything that can prepare and commit. A coordinator is placed over them, and then the process proceeds in two rounds.

Prepare phase (voting phase). The coordinator sends a request to all participants to prepare for commit. Each participant locally performs their part of the transaction (for example, makes the necessary changes to their database) but doesn't commit them, marking them as "ready to commit" (in databases, this usually means writing the changes to the log and locking resources). After this, the participant responds to the coordinator with either "ready" (Yes) if their stage has been successful and they are ready to commit, or "cannot" (No) if an error has occurred and they are unable to commit. All participants essentially vote "yes" or "no" on a shared commit.

Commit phase. The coordinator collects responses. If all participants respond "yes," the coordinator sends a Commit command to all participants. Each participant receives this command and commits their changes (permanently applies them to their system). If at least one participant rejects the request ("No") or fails to respond due to a failure, the coordinator sends a Rollback command to all participants who were ready. Either all participants commit or all participants roll back — the protocol allows no third outcome.

After completing the second phase, the coordinator can notify the initiator (e.g., the application that started the transaction) that the transaction was successfully completed or rolled back.

A full implementation of the coordinator would take several hundred lines, but the entire protocol rests on three points. We'll show them here.

The first thing the coordinator does after collecting votes:

// Phase 1: voting
boolean allReady = participants.stream().allMatch(p -> p.prepare(txId));
Decision decision = allReady ? COMMIT : ABORT;

// The key line of the whole protocol: the decision hits the log BEFORE it is broadcast.
// If the coordinator dies right after it, on startup it will read the log
// and drive the transaction to completion. Without fsync this is pointless.
log.writeDecision(txId, decision);

// Phase 2: the participants have already voted "yes" and must comply.
// An error at this stage is no reason to roll back - it is a reason to retry.
for (Participant p : participants) {
    retryUntilSuccess(() -> p.apply(txId, decision));
}

A naive implementation broadcasts the decision immediately and gives up at the first error. Here, the decision is recorded in the log before anyone else knows about it: this record is what makes the protocol recoverable. Broadcasting is retried until it succeeds — after the voting phase, a participant's failure no longer changes anything; the transaction must complete as the coordinator decided.

What happens on the participant's side:

boolean prepare(TxId id) {
    // the only moment when "no" is still an option
    if (!canCommit()) return false;
    // changes are on disk, but visible to no one
    writeUndoRedoLog(id);
    // these are the locks that will hang
    lockRows(id);
    // the state survives a restart of the participant itself
    markInDoubt(id);
    // after this, backing out is no longer possible
    return true;
}

The markInDoubt line is the very same suspended transaction discussed above. The participant has written the changes, holds the locks, and is physically unable to make a decision on their own: they don't know how the vote went for everyone else. Any XA-compliant DBMS displays such transactions in a system view, and they must be resolved either by the returning coordinator or by an administrator by hand. Manual resolution is called a heuristic decision and is dangerous precisely because the administrator might choose something different from the coordinator: then one part of the system will commit while another rolls back — the very violation of atomicity that the protocol was designed to prevent.

Recovery: three lines that explain why the log was needed.

// On coordinator startup
for (Tx tx : log.unfinished()) {
    // no decision in the log - abort
    Decision d = tx.decision().orElse(ABORT);
    for (Participant p : tx.participants()) retryUntilSuccess(() -> p.apply(tx.id, d));
}

The logic is simple: if the decision made it into the log, we play it out; if not, the coordinator died before the point of no return, and it's safe to abort. Inside retryUntilSuccess live timeouts, exponential backoff, and the attempt limit after which a human is paged. I've left those out of the listing, but that's where the protocol's real price hides.

This also reveals 2PC's real weakness. It's not the two phases or the locks, but the fact that the decision log exists in a single copy. While the coordinator is down, the participants wait and can't do anything. Distributed DBMSs solve exactly this problem by replicating the coordinator's log via Raft or Paxos: there are now multiple copies of the decision, the loss of a node is no longer fatal, and the protocol turns from unreliable to workable.

In real systems, the role of participants is performed by resource managers: each database can write changes to its log and confirm its readiness upon a prepare request, and complete the transaction upon a commit request. The coordinator is a transaction manager embedded in the application or the runtime environment.

2PC applications and tool support

The 2PC protocol was widely used in traditional enterprise applications, especially before the microservices era. Typical places where 2PC is encountered:

Distributed relational databases — for example, a transaction affects two different databases (two DBMSs or two connections). Java provides JTA (Java Transaction API) and XA drivers for this purpose; the coordinator (Narayana, Atomikos, etc.) ensures a two-phase commit between the databases.

A database plus a message queue — the classic use case of writing data to the database and sending a message without ending up with "written to the database but the message was lost," or the reverse. If the broker can be an XA resource, it can be included in a single global transaction with the database. Not everyone can do this: XA is supported by ActiveMQ (Classic and Artemis), IBM MQ, Oracle AQ, and other JMS providers implementing XAConnectionFactory. However, the two most popular brokers today, Kafka and RabbitMQ, are not XA resources. Kafka has its own producer transactions, but they are atomic only within Kafka and have no knowledge of your database. RabbitMQ has AMQP channel transactions, which are also not XA. For a typical modern stack, this path is simply a dead end, and this is where the Outbox pattern, which we'll return to below, comes from.

Classic monolithic systems with multiple resources — a transaction updating several subsystems (for example, two databases, or a database and a file system) could also be coordinated via 2PC.

Implementing 2PC requires support from all participants. Each resource must have a prepare/commit interface. In the world of relational databases, this is the XA standard; NoSQL stores or custom services often lack such support. Therefore, in pure microservices, where services are heterogeneous, implementing 2PC yourself is difficult — you either need to write an adapter layer for each one (so that the services can accept prepare/commit commands) or limit yourself to resources that already support XA. There are ready-made transaction managers (coordinators) — the aforementioned Atomikos, Narayana, Bitronix, and others — that can be embedded into an application and configured with resources to participate in a global transaction. However, this entails significant limitations, which are discussed below.

Why it's almost never used between services

Locks are held for the entire time between the phases. A participant who responded "ready" holds the rows locked until a command arrives and cannot release them on its own. If there are five participants, the slowest one sets the pace: while one is thinking, everyone is holding resources. The scaling ceiling here is set by the number of participants and the spread of their response times.

Availability degrades multiplicatively — we've already done that arithmetic above. The practical consequence: if one participant is unavailable, the whole operation fails. In this situation, a saga can wait until the service returns, or go into compensation. 2PC has no such choice: it either collects all the votes or cancels the transaction.

And coupling. All participants are required to obey a common coordinator and a common protocol — exactly what you were escaping when you split the monolith into services. The result is a distributed monolith at the transaction level: the services are separate, but they can only commit together.

2PC remains useful where the participants are few and homogeneous. A typical example is transferring money between accounts stored in the same sharded database but on different shards. A saga is bad here not because of code complexity but because of semantics: there's a window between the debit and the credit when no funds exist in either account, and a report taken at that moment will show an incorrect system-wide total. And the conditions that ruin 2PC between microservices aren't met here: there are exactly two participants, they're homogeneous, they're in the same cluster, and the lock is held for microseconds. That's why sharding layers like Citus perform cross-shard writes using a two-phase commit via the standard PREPARE TRANSACTION and COMMIT PREPARED statements in PostgreSQL.

And that's just the application layer. Within the storage layer, 2PC is alive and well: Google Spanner, CockroachDB, TiDB, and YugabyteDB perform distributed transactions using a two-phase commit, simply on top of consensus. This fixes the protocol's core problem: classic 2PC blocks when the coordinator holding the single copy of the decision goes down. If the coordinator's log is replicated via Raft or Paxos, the decision survives the loss of a node, and participants can always learn the outcome. Add homogeneous participants, a single cluster, sub-millisecond RTT, and short locks — and a protocol that was unbearable between microservices turns out to be perfectly workable.

Hence the practical conclusion: don't write 2PC yourself — get it with the database. If the invariant truly requires strict atomicity, it's cheaper to put related data in a single distributed DBMS that handles 2PC for you than to build an XA coordinator on top of your own services.

Comparing Saga and 2PC

They share the same goal, but differ in four areas.

	Saga	2PC
What's visible from the outside while the operation runs	The intermediate state is visible to everyone: the order is created, the money isn't charged yet. This has to be described in the service contract.	Nothing. Only the final result crosses the boundary.
What's required of the participants	Nothing beyond local transactions. Any service, any database.	XA support. Kafka, RabbitMQ, and most REST services cannot be participants.
What happens on failure	Compensations, which you write yourself. There is no "without a trace" rollback; intermediate effects stay in the history.	Centralized rollback. If the failure happens after the prepare phase, the transaction hangs until the coordinator or an administrator comes back.
Where it fits	Long business processes: orders, billing, bookings. Anywhere steps are compensatable and a second of latency hurts no one.	Participants are homogeneous, few, and in the same cluster: cross-shard writes, distributed DBMSs, several databases under one XA coordinator.

Alternative approaches and patterns

Saga and 2PC are poles apart, but there are intermediate options between them, and these are the most common in practice.

TCC (Try-Confirm-Cancel)

A customer books a flight and a hotel together. The ticket is issued, but the rooms are sold out — and now they have to return the already-sold ticket. TCC (Try-Confirm-Cancel) addresses these situations by not selling anything until the last minute: first, everything is reserved (Try), and only when all parties have reservations does a confirmation (Confirm) or cancellation (Cancel) follow.

How TCC works. Let's imagine booking an air ticket and a hotel together. In the TCC model, the steps are as follows:

Try. The ticket service receives a booking request and makes a preliminary reservation (without finalizing the sale). At the same time, the hotel service reserves the room. These "Try" steps are performed for all participating services — they allocate the necessary resources, marking them as busy, but the transaction is not yet considered complete.

Confirm. If all services have successfully completed the Try step and are ready to complete the transaction, a Confirm command is sent to each service — tickets are issued permanently, and hotel reservations are confirmed. Each service makes the final change.

Cancel. If any of the Try steps fails (or one of the services responds that it cannot complete the transaction), a Cancel command is sent to all services that have already completed the Try step — they cancel any previously made reservations (free up seats, do not charge funds, and so on).

Externally, this appears atomic: either all Confirms are successful, or everything reserved is canceled with Cancel. This is similar to 2PC (Try is analogous to prepare, Confirm/Cancel to commit/rollback), but the difference isn't in the presence of a coordinator: in real implementations, one exists — in Seata, the same Transaction Coordinator is responsible for TCC mode as for the other modes. The difference is who implements the phases and what is held between them.

Who implements the phases. In XA/2PC, prepare and commit are implemented by the resource manager — the DBMS itself; the application is unaware of them. In TCC, the developer writes three methods: Try, Confirm, and Cancel — regular service methods called over regular HTTP or gRPC. This is the main advantage: any service that can serve three HTTP methods can become a participant.

What locks are held. In 2PC, a physical row lock is held between prepare and commit, held by a transaction opened over the network. In TCC, the Try step is committed locally and immediately — the "lock" becomes semantic: a reservation row with a status and a lifetime. There are no physical locks between services, and a stuck reservation is released by timeout rather than waiting for an administrator.

Compared to Saga, TCC is a saga whose intermediate effects are not visible from the outside: before Confirm, the order isn't placed and the money isn't debited, only a reservation exists.

TCC is a compromise: the strictness of 2PC, achieved by the application and without XA. Here's what you pay for it.

The service contract triples in size. Instead of one method there are three — Try, Confirm, Cancel — and the two-phase nature has to be dragged into the local logic. The hotel booking service now has to hold a room reserved, mark it as occupied and paid, and cancel a reservation: three states where there used to be two.

Reservations have to be released by someone. While the client is deciding whether to confirm, the room is unavailable to others. If a Confirm doesn't arrive within the agreed-upon time, Cancel must execute on its own, otherwise phantom reservations accumulate until someone notices. A Cancel that never arrives is a phantom too, so a background process is needed to find expired reservations and release them itself.

Idempotency — the same as in a saga, only now for three methods instead of two. A repeated Confirm shouldn't break an already confirmed reservation, and a repeated Cancel shouldn't break an already canceled one.

On the upside, failure is visible early: if a resource is unavailable during Try, no actual change occurs at all, and there's nothing to compensate for.

The main limitation is that not everything can be reserved. Bookings, holding funds on a card, capacity allocation — yes. Sending an email or writing to a log can't be reserved; those have to be handled by a saga with compensation, or ignored on cancellation.

In practice, TCC is implemented either by hand or using ready-made solutions: Seata has a TCC mode (alongside AT, XA, and SAGA), as does ByteTCC, while Atomikos supports it in the commercial ExtremeTransactions, under the same Try-Confirm/Cancel name. There is no established synonym for this pattern; in the literature, it is simply called TCC.

Outbox Pattern (Transactional Outbox)

The dual-write problem is one of the most common in microservices: how can we guarantee that an action in a local database and the sending of an event or message to another service occur atomically? For example, an Order Service has saved an order in its database and needs to send an OrderCreated event to Kafka for other services. If the write to the database succeeds but sending the message fails, the data has already changed, and other services won't know about it. Conversely, if the message is sent but the database isn't saved, other services will learn about an order that doesn't exist. Classic 2PC between the database and the broker is often unavailable. The Outbox pattern solves this problem by ensuring that the database and the message stay in sync.

The essence of the Outbox pattern: instead of sending a message directly, the service first stores the message information in a special Outbox table in its local database within the same transaction as the main data. Then, a separate process or thread reads this table and actually sends the messages to external recipients. The algorithm:

Local transaction. A request (for example, to create an order) arrives at the service. The service opens a transaction to its database. Within it, it performs the usual changes (creates an order record) and simultaneously inserts a record into the Outbox table — for example, a JSON payload describing the OrderCreated event that needs to be sent, plus a "new" status. The transaction is then committed. If for some reason the database write didn't go through, neither the order nor the event ends up in the Outbox. If the commit succeeds, both the order and the message record are durably in the database.

Sending from the Outbox. A separate component — let's call it the Outbox Processor — periodically reads new records from the Outbox table. For each record, it performs the actual send to the broker or calls an external service. After a successful send, it marks the Outbox record as sent. This operation is also transactional, locally within the database.

An order and an event are linked by a single commit: either both exist, or neither does. Even if sending to the broker is temporarily impossible, the record sits durably in the database and will wait for the next attempt. The database and the event stream can no longer diverge.

The technique comes down to a single trade: you get a delivery guarantee and pay for it in duplicates. The event will definitely be sent — it's in the database and will wait until the broker recovers. But it will be sent at least once, not exactly once: if the processor crashes after sending and before writing the "sent" mark, it will send again on restart. This means deduplication by message ID is mandatory on the receiving end — either an Inbox table or a unique index. Strict exactly-once isn't achievable here, and it's not worth relying on.

The rest is operational overhead. The table grows, it needs to be cleaned up, and its indexes watched: you've created your own little queue inside the service, with all the responsibilities of a queue. Sending happens with a delay of milliseconds to seconds, depending on the polling frequency or CDC settings.

You don't have to write your own processor: Debezium reads the transaction log and publishes events to Kafka. The price is a separate piece of infrastructure: Kafka Connect with a connector, access to the WAL or binlog with replication rights, and for the outbox table, usually an SMT router that fans records out into topics. "No code" here doesn't mean "no operations."

Despite these drawbacks, the Outbox pattern has become the de facto standard for building reliable asynchronous integrations between microservices. It complements Saga particularly well: for example, one service uses Outbox to publish an event that triggers the next Saga step in another service, ensuring that no step is lost.

Other techniques: eventual consistency, transactional messaging

A few more things that come up in any discussion of distributed transactions.

BASE model and eventual consistency. The counterpart to ACID. Basically Available, Soft state, Eventually consistent — a principle widely used in distributed systems: the system is always available for operation, but allows data to temporarily diverge, with consistency achieved "eventually." A saga is a special case of eventual consistency. Other examples: Event Sourcing (where state is computed from a stream of events), CQRS (separate command and query models, synchronized through events). In microservices, immediate consistency isn't always necessary; it's often enough that all services converge on the same data within a second or two.

Transactional messaging. A term for different ways to ensure atomicity between sending a message and changing state. Outbox is one of them. Another approach is to use the broker itself as a state store: send a command to a topic and consider the operation complete only after the other side confirms processing. Brokers also have their own transactions, but, as mentioned, they are atomic within the broker and don't cover your database, so they don't solve the dual-write problem on their own. There's a pattern called Transactional Inbox/Outbox, where incoming messages on the receiving end are also written to a local table and processed atomically with the service's local transaction — essentially an outbox on both ends.

Commercial distributed transactions. Historically, there have been products that enable two-phase commits between disparate systems. For example, Atomikos is a popular transaction manager for Java that allows multiple resources (databases, queues) to be included in a single JTA transaction. IBM MQ and IBM TX Series are examples of industrial-strength distributed transaction solutions. These solutions work, but, as noted, are rarely used in microservices due to their complexity and the requirements they place on participants.

Three-Phase Commit (3PC). Adds a third phase between voting and committing to allow participants to complete the transaction without a coordinator. It's non-blocking only in a model with fail-stop nodes and without network partitions. Under a partition, 3PC breaks not availability but correctness — two parts of the cluster can make opposing decisions. That's why it didn't catch on: the coordinator was made fault-tolerant by replicating the decision instead (Paxos Commit, Gray and Lamport, 2006).

An infrastructure-level solution. Related data is stored in a single distributed DBMS, which performs the distributed commit itself (see above on 2PC over consensus). Consistency is provided by the database, and this is perhaps the cheapest way to achieve strict atomicity where it's truly needed.

Practical recommendations: how to choose an approach

Here's what I check, in order, when the first operation spanning two services appears in a project.

Do there really need to be two services? The test isn't "is it difficult to implement" but "does one of these services ever change without the other?" If it never does, the boundary is drawn wrong, and a distributed transaction here only treats the symptom. Merging the services or duplicating the data is cheaper than building a protocol on top of a decomposition error.

What does the client get back? This question decides more than all the others. If the API can respond "accepted, ask for the status later," then a saga fits, and from there it's all about compensations. If the client needs a final answer within the same request, the saga is out: you'd have to expose an intermediate state, and there's nothing to expose. That leaves TCC with reservations, or moving the related data into a single database.

Who are the participants? If Kafka, RabbitMQ, or someone else's HTTP service is among them, the question of 2PC is off the table — XA isn't there and won't be: that leaves Outbox for delivery and a saga for the process. If the participants are your own, homogeneous, and in the same cluster, it's worth checking whether the problem can be solved by moving to a distributed DBMS that will do the two-phase commit for you.

Where is the point of no return? Go through the steps and find the first one that can't be undone automatically. If it turns out to be the second of five, the saga is nearly useless — there'll be nothing to roll back. Then either the step is split into two phases, like authorization and capture, or the order of steps changes, or the operation isn't suited to a saga at all.

How much does an error cost? This question comes last, because the answer to it can override the previous four. The mechanism is chosen by the cost of the operation, not one for the entire product — that was covered above in the countermeasures. And if the operation is expensive and irreversible, it's cheaper to take it out of automation entirely and require human confirmation.

What's required regardless

Idempotency and a correlation identifier are mandatory in any case, and both were covered above. There's a third thing that wasn't: every distributed operation needs a deadline and an escalation target. A saga that can neither complete nor roll back must not hang forever, and must not disappear quietly — after N attempts it has to land in a human's review queue. This is the most boring piece of the scaffolding and the first one people forget to write, because it never fires before production.

A distributed transaction is visible from the outside. The client receives "payment processing" instead of "paid" and should be able to ask for the status later. This decision is made when designing the API, not when choosing a library. 2PC hides the complexity in the infrastructure and works as long as the participants are homogeneous and sit close together. Saga moves it into the code and the business logic: you have to write more, but you can see what happens on failure and who is responsible for it. The worst option is the third one — to assume there's no complexity, and find it in production when the money has been debited and the order hasn't been created.

Paxos, Raft, and Zab Consensus Algorithms in Distributed Systems

Timofei Ivankov — Sun, 02 Aug 2026 14:14:06 +0000

Introduction

Consensus in a distributed system is the agreement of multiple nodes on a common decision or order of operations, which persists despite node failures and network latency. The problem seems trivial until it becomes clear that a node can't distinguish a downed neighbor from a slow one, and the network is entitled to deliver a message an hour after sending.

Paxos, Raft, and Zab diverge in almost everything: the roles of nodes, the method for electing a leader, and what happens after a failure. They share one requirement: decisions must be made by a group of nodes. Any two such groups must overlap, and the simplest way to guarantee this is a majority. Everything else stems from this requirement, including the differences between etcd, ZooKeeper, and Consul, which, while sharing a common foundation, provide different read guarantees for different clients.

Paxos: what safety costs

Paxos is a basic consensus algorithm that guarantees a single decision (a single value) in an asynchronous network even if nodes fail, as long as at least a majority remains operational. Paxos operates with three node roles-proposers, acceptors, and learners-which can coexist on the same servers. Proposers propose values, acceptors vote on them, and learners learn the outcome of their decisions. A key property of Paxos is safety: it guarantees that two different nodes will not reach different decisions, even if messages are lost or nodes are rebooted. However, liveness is only possible when a quorum of nodes is available and there is no endless duel between proposers.

This limitation has a rigorous justification. The FLP theorem (Fischer, Lynch, Paterson, 1985) proves that in a fully asynchronous network, where there is no upper bound on message latency or node speed, a deterministic consensus algorithm cannot guarantee termination even if a single node fails. This is due to indistinguishability: the receiver is unable to distinguish a downed node from a very slow one, and any finite latency is arbitrary.

Paxos overcomes this by separating its guarantees. Safety holds unconditionally, regardless of delays and message losses: two nodes will never commit to different decisions. Progress, however, is guaranteed only when the network behaves sufficiently predictably—that is, in a model of partial synchrony. This is the origin of dueling proposers: two nodes can take turns bidding on each other with increasingly higher numbers, each time wiping out the other's work. The protocol itself doesn't protect against this, in practice, the problem is solved with a dedicated leader and a randomized delay before retrying, so that proposers don't start synchronously.

Partial synchrony isn't the only way out of FLP. A second option is to abandon determinism. Randomized protocols like Ben-Or flip a coin where a deterministic algorithm stalls and terminate with probability 1, but with no upper bound on the number of rounds. The "almost certainly someday" guarantee doesn't hold well for a system that's expected to respond within tens of milliseconds, so timeouts have become the norm in general-purpose infrastructure. Randomization remains in places where timeouts can't be trusted in principle—in Byzantine protocols.

This distinction holds true for all three algorithms in the article. Neither Paxos, Raft, nor Zab overcome FLP, they merely choose differently where to place timeouts and how quickly to converge to a single leader.

Phases of the Paxos and Multi-Paxos protocol

Basic Paxos (sometimes called Single-Decree Paxos) consists of a two-phase protocol with message exchanges between proposers and acceptors. These phases are often referred to as Prepare and Accept:

Phase 1: Prepare. The proposer selects a unique proposal number n that is the highest of all previously used proposals and broadcasts a Prepare(n) request to all acceptors. An acceptor, upon receiving a Prepare with a number higher than any previously seen, responds with a Promise(n) and promises not to accept proposals with a lower number. It also informs the proposer which v_a value with which n_a number it has already accepted, if any.

Phase 2: Accept. After receiving responses from a majority of acceptors (a quorum) in the Prepare phase, the proposer determines the value to accept and has no choice. If at least one acceptor in the quorum has announced a previously accepted value, the proposer must accept the one with the highest n_a number. They may propose their own value only if the quorum has not yet accepted anything. Deviating from this rule breaks safety: a proposal with a higher number will overwrite an already chosen solution. Uniqueness of a solution follows from the intersection of quorums: any two majorities have at least one common acceptor, and an acceptor cannot accept two different values without breaking their promise. If a value has already been chosen, any proposal with a higher number must propose it—this prevents the solution from being lost during a leader change. Therefore, the value selection rule in phase 2 is formulated as an obligation, not an optimization.

For a sequence of decisions (e.g., writing a command log), basic Paxos is applied multiple times for different "slot numbers" in the log. In practice, an optimized Multi-Paxos mode is used, in which, after the initial round, one node acts as the leader (coordinator) for all subsequent writes until a failure occurs. The leader, upon receiving client transactions, acts as the proposer for new values and, thanks to the trust of the acceptors, can skip the Prepare phase for each value, sending Accept requests directly (meaning decisions are made faster, similar to continuous operation with a single coordinator). This achieves efficiency similar to single-leader replication: Prepare phase messages are infrequent until the leader changes.

Paxos does not have a separate election mechanism, and this is a fundamental difference from Raft and Zab. Leadership here occurs de facto: the proposer whose proposal with the highest number has passed the Prepare phase with a quorum becomes the leader. Nothing prevents two nodes from simultaneously claiming the leadership—safety is not affected, only progress. Therefore, all practical implementations of Multi-Paxos build on top of the protocol: they appoint a coordinator administratively, elect it via a failure detector or a separate leadership protocol, and add a lease with a timeout. The standard doesn't specify what exactly to do, and each system decides for itself. This unaddressed gap is the commonly cited "Paxos complexity": the algorithm itself is short, but everything needed around it for a working system is unspecified.

During a network partition, a Paxos system maintains consistency at the cost of availability: a minority of the cluster stops moving, but doesn't write anything inconsistent. The gap between the strict core and the unspecified framework gave birth to Raft—an attempt to describe not only the algorithm, but also everything needed around it for a working system.

Failure Handling and Recovery in Paxos

The Paxos algorithm is resilient to node failures: as long as at least a majority of acceptors are operational, a new value can be selected. If the current leader (the Multi-Paxos coordinator) fails or loses contact, other proposers can initiate new Prepare rounds with a higher proposal number, effectively electing a new leader. Due to the quorum system, the old and new leaders can never each secure a majority of votes simultaneously, so consistency is not compromised. Paxos tolerates long stalls (for example, if less than half the nodes are available, progress is suspended, but the previously reached decision is not reversed). Upon recovery, nodes synchronize their logs based on the accepted values: already committed operations will be delivered from learners or the leader. The protocol requires each acceptor to store on disk its last promised number n_p and the accepted value v_a with the number n_a. This ensures that a node, upon recovery, does not break previously made promises not to vote for "old" proposals. Thus, Paxos ensures data persistence: neither node failures nor arbitrary message delays lead to inconsistencies – the system either extends the choice time or pauses until a quorum is restored, but does not make incorrect decisions.

Paxos pseudocode example

Below is a simplified pseudocode of how the key Paxos roles – proposer and acceptor – work, demonstrating the described phases of the algorithm (based on the classical description):

function proposer(node_id, value):
    round = last_round_seen + 1
    n = (round, node_id)

    send Prepare(n) to all acceptors

    # --- Phase 1 ---
    wait for Promise/Nack with timeout
    if received Nack(n_p_other):
        last_round_seen = max(last_round_seen, n_p_other.round)
        # otherwise proposers duel forever
        sleep(randomized_backoff)
        retry
    if not quorum_of(Promise):
        sleep(randomized_backoff); retry

    # --- Value selection ---
    # Must reuse an already accepted value, otherwise safety is lost.
    accepted = [r for r in promises if r.v_a is not None]
    if accepted:
        # compare PAIRS, not plain numbers
        v = v_a of response with maximal n_a
    else:
        v = value

    # --- Phase 2 ---
    send Accept(n, v) to all acceptors
    wait for Accepted/Nack with timeout
    if quorum_of(Accepted(n)):
        send Decided(v) to learners
    else:
        sleep(randomized_backoff); retry with higher round

# Acceptor - keeps persistent state n_p, n_a, v_a
on receive Prepare(n):
    # proposal is newer than anything seen before
    if n > n_p:
        n_p = n
        fsync(n_p)
        # report our last accepted value (if any)
        reply Promise(n, n_a, v_a)
    else:
        reply Nack(n_p)

on receive Accept(n, v):
    # proposal is not stale
    if n >= n_p:
        n_p = n
        n_a = n
        # accept the new value under number n
        v_a = v
        fsync(n_p, n_a, v_a)
        # acknowledge acceptance
        reply Accepted(n)
    else:
        reply Nack(n_p)

This protocol ensures that an acceptor who promised not to accept old numbers will not accept a value from the old leader after seeing a newer proposal. And the proposer, having learned of previously accepted values (via v_a), resubmits them to avoid losing what has already been committed. Ultimately, unanimity is achieved: once a majority of acceptors accept a value, it becomes the decision.

Raft: Same result, but with a clear leader

Raft (Diego Ongaro, John Ousterhout, 2014) was created as a simpler, more understandable alternative consensus algorithm that provides the same properties as Paxos. Raft also belongs to the class of iterative consensus algorithms for a replicated log—that is, it maintains a replicated state machine: all nodes apply the same commands in the same order, ensuring that their state remains consistent. Unlike classic Paxos, Raft explicitly separates subproblems: (1) leader election, (2) log replication, and (3) safety. The core idea of Raft is to always have an explicitly elected leader through which all written changes flow, simplifying the protocol's understanding.

In a Raft cluster, each node can be in one of three states: Follower, Candidate, or Leader. In normal mode, there is a single leader, and the remaining nodes are followers, passively replicating its log. If the leader fails or contact with it is lost, the nodes proceed to elect a new leader. Below is a diagram of node states in Raft and the transitions between them during leader election:

Leader selection in Raft

Raft achieves consensus leader election through rounds of voting called terms. Terms are numbered and stored on each node. At the beginning of the process, all nodes are Followers and have no leader. Each Follower starts a randomly assigned timer (the election timeout). If the timeout expires without receiving a heartbeat message from the leader, the node assumes there is no leader and transitions to the Candidate state, increments its term, and begins electing a new leader. The candidate votes for itself and broadcasts a RequestVote to all other nodes, specifying its term and the index of the last entry in its log. Each node (Follower) receiving this request decides whether to grant or reject the vote, according to the following rules:

The voting Follower compares the candidate's term with its current term. If the candidate's term is lower (outdated), the vote is rejected. If the candidate's term is not lower, the Follower updates its current term and can vote.

Each Follower has the right to vote for only one candidate within a single term (it remembers votedFor = candidateId). Repeated requests from other candidates in the same term are rejected.

A Follower also checks the candidate's "log freshness": it votes only if the candidate's log is at least as up-to-date as its own (based on the index and term of the last entry). This ensures that the candidate with the most advanced log is elected, facilitating subsequent log synchronization.

The candidate that receives the votes of a majority of nodes (quorum > N/2) wins and becomes the new leader (transitions to the Leader state). From this point on, it sends periodic Heartbeat messages (empty AppendEntries RPCs) to all other nodes to notify them of its leadership and prevent new elections. The election can fail in two cases. Either the candidate fails to secure a majority—for example, if the votes are split between several candidates—in which case it waits for a new timeout and begins the next term. Either it has received AppendEntries from the current leader with a term equal to or greater than its own, in which case it returns to Follower and accepts that leader. The use of randomized timeouts prevents repeated split votes—usually, one node will initiate elections first. Raft elections are fast (within about two timeouts) and guarantee that no two leaders can emerge in a single term at any given time (thanks to majority voting and the one-vote rule). Each new term is logged, records of previous terms may be incomplete, but Raft guarantees that they will not affect consistency.

Log replication and write commit

Once a leader is elected, all client requests for state changes flow through it. The leader accepts, for example, a command to "write value X," adds the corresponding entry to its local log, and assigns it the current term and next ordinal index. The leader then sends AppendEntries RPCs to its followers with the new entries. In the AppendEntries request, the leader specifies the previous index and term to ensure consistency. Each follower receiving such a request checks: if its log does not have an entry with the specified previous index and term (i.e., the local log is lagging or contains a conflicting end), the follower rejects the AppendEntries request. This mechanism, along with the log freshness check during voting, ensures log convergence: sooner or later, the logs of all nodes will become identical sequences of commands.

Once a record has replicated to a majority of nodes (the leader has received AppendEntries confirmations from the quorum), the leader can mark it as committed, but only if this record belongs to its current term. Replica count alone does not grant the right to commit. A record inherited from the previous leader cannot be committed by the replica count, even if it is on a majority of nodes, it can still be overwritten by a future leader. Such records are committed indirectly: as soon as the leader has committed at least one record of its term, the entire preceding log prefix is considered committed automatically, as follows from the log-matching property. Therefore, in practice, a new leader immediately after being elected writes an empty no-op record to the log: this unlocks the commit of the inherited tail without waiting for client operations.

A committed record is considered finally applied, the leader executes the command in its state machine and returns the result to the client. In each AppendEntries, the leader also transmits the index of the last entry it committed (commitIndex). After receiving this message, the follower also marks the corresponding entries as committed. Therefore, the moment a record becomes visible varies across nodes, and replication to the majority by itself does not provide linearizability of reads. Raft prioritizes consistency over availability: if there are insufficient nodes for quorum (for example, the network is split and the leader is disconnected from the majority), the leader will be unable to advance commitIndex—the system suspends change processing, preserving the previously achieved consistency. In CAP terms, this is a CP system, but the framework itself is rather crude: it describes behavior only during a partition, and a partition is a rare event. The rest of the time, when the network is healthy, the tradeoff remains, it simply swings between latency and consistency: a quorum of three data centers is consistent, but each write is paid for by an interregional round-trip. PACELC captures exactly this: when partitioning (P), choose between A and C, otherwise (E), choose between latency (L) and consistency (C). The article goes on to explain what decides the other half: ReadIndex versus lease read, sync in ZooKeeper versus reading from a local replica, and Cosmos DB consistency levels.

Reads can be served without appending to the log, but not naively "from the leader": an isolated leader doesn't know it has already been deposed and will return stale data. Linearizable reads require ReadIndex (leadership confirmation via a heartbeat round before replying) or lease read based on limited clock drift, reading directly from a follower is only acceptable if lag is acceptable.

Raft uses a more rigid leadership model than Paxos: the leader completely controls the command flow, and followers do not participate in proposing new values. This makes it easier to understand – the system boils down to classic master-slave replication, supplemented by safe leader reelection in the event of a failure. Raft's log matching property: if two records on different nodes have the same index and term, then the entire log prefix up to that index is guaranteed to match. This is achieved by rolling back conflicting entries on followers: if a record on a follower doesn't match the leader's record in terms of term (meaning that it was previously written under the leadership of another leader who failed to commit it), the leader forces the follower to delete this "orphaned" part of the log and replace it with records from its own log. This restores log consistency.

A consistent log alone is insufficient for correct operation, and this is a common oversight in home-brew implementations. A client sends a command, the leader commits it, and crashes before responding. The client times out and repeats the request to the new leader, and the command is applied a second time. This is harmless for "set X = 5," but not for "debit $100". Raft solves this at the state machine level, not the log level: each client receives a unique identifier and numbers its requests. The state machine stores the number of the last command applied and its result for each client. A repeat request with a number already seen is not applied, the saved response is returned. The log contains both copies of the command, and duplicates are pruned upon application. Session records must be cleared by timeout, otherwise the table grows indefinitely, and expired sessions must be rejected with an explicit error, not silently.

Crash Handling and Recovery in Raft

Raft is designed to remain safe (not to contradict already committed records) even in the face of arbitrary node shutdowns, network delays, or partitions. Fault tolerance is achieved as long as a majority of nodes is operational, i.e., ⌊N / 2⌋ + 1. This is the minimum required condition: two disjoint sets of nodes should not be able to independently make decisions, and the intersection of any two majorities is guaranteed to be non-empty. If the leader fails, detection occurs via a timeout. Followers initiate the election of a new leader, as described. Each node stores the currentTerm and votedFor values on disk, as well as its entire log. This means that a node rebooting does not "forget" who it voted for in a given term, nor does it lose its log records. As noted above, the leader does not commit the records of previous terms based on the number of replicas—only indirectly, by committing the record of its own term. This rule, along with the log freshness check during voting, ensures that the new leader always contains all the records committed by its predecessors. This ensures that the new leader continues from a correct state.

If a node falls behind (for example, due to being unavailable for a long time), the log catch-up mechanism allows it to catch up. The leader specifies prevLogIndex/prevLogTerm in AppendEntries messages for checking. If a lagging follower is desynchronized, it will receive an AppendEntries rejection. The leader will decrease prevLogIndex until it finds a common point with the follower's log. If the leader has already discarded the necessary records while compacting its own log, finding a common point is impossible, and then InstallSnapshot is sent instead of AppendEntries (see below for log compaction).

Raft includes a built-in log compaction mechanism. Snapshots are created independently by each node: they are not a leader operation and are not a consensus decision. A node takes the current state of its machine, writes it to disk along with the index and term of the last record included in the snapshot, and then discards the entire preceding log prefix. The node itself chooses the moment to take the snapshot, usually based on the log size. This prevents unbounded log growth and speeds up node restarts: it restores from its snapshot rather than replaying the log from scratch.

If the records needed by the follower have already been discarded, the leader sends it an InstallSnapshot RPC: its entire snapshot (in chunks, if necessary), after which the follower replaces its state with it and continues receiving regular AppendEntries.

Raft also defines a procedure for safely changing cluster membership through joint consensus: it introduces a transitional configuration, C_old,new, in which each decision requires a majority in both the old and new membership—not a larger majority over the union of the two, but two independent majorities. This eliminates the possibility of two leaders emerging during the transition. After committing C_old,new, the leader commits C_new, and the old configuration is removed from play. In practice, the simpler mechanism from Ongaro's dissertation is more commonly used: adding or removing exactly one node at a time, in which the majorities of the old and new configurations overlap automatically and a transition configuration is not required.

As a result, Raft provides developers with a relatively simple model: a single leader replicating commands to followers. It guarantees the linearizability of operations (when using ReadIndex or lease read) and maintains availability in the event of failure of up to ⌊(N−1)/2⌋ nodes, meaning a 3-node cluster survives the failure of one, a 5-node cluster survives the failure of two, and a 7-node cluster survives the failure of three.

This leads to a practical consequence: an even number of nodes does not provide a gain. A 4-node cluster requires a quorum of 3 and therefore survives the failure of only one node—exactly like a 3-node cluster, but each write must reach more replicas, and the probability of at least one node failing is higher. Therefore, Raft clusters are almost always deployed with odd numbers: 3, 5, or 7 nodes. More than 7 are rarely used: fault tolerance increases slowly, and commit latency is determined by the slowest quorum node. Raft implementations demonstrate throughput in the order of thousands to tens of thousands of operations per second per cluster, depending on the workload and hardware. The performance is usually limited by the fsync speed on the leader disk and the network latency to quorum, rather than by the properties of the algorithm itself.

Raft implementation example (code snippet)

class RaftNode:
    def __init__(self, node_id, storage):
        self.node_id = node_id
        self.storage = storage
        # --- persistent state ---
        self.current_term = 0
        self.voted_for = None
        self.log = []
        # --- volatile ---
        self.state = "follower"

    def _persist(self):
        self.storage.save(current_term=self.current_term,
                          voted_for=self.voted_for)
        self.storage.fsync()

    def _last_log(self):
        if not self.log:
            return (0, 0)
        return (len(self.log), self.log[-1].term)

    def _step_down(self, term):
        self.current_term = term
        self.voted_for = None
        self.state = "follower"

    def _log_is_up_to_date(self, cand_index, cand_term):
        my_index, my_term = self._last_log()
        if cand_term != my_term:
            return cand_term > my_term
        return cand_index >= my_index

    # ---------- RPC ----------

    def on_request_vote(self, term, candidate_id, last_log_index, last_log_term):
        term_changed = False

        # 1. Stale term - reject. Always return our own term
        #    so the candidate can step down on its own.
        if term < self.current_term:
            return (self.current_term, False)

        # 2. Term is higher than ours - adopt it and RESET voted_for
        #    BEFORE checking whether we may vote. Skipping the reset here
        #    is the classic bug: the node would refuse every candidate
        #    of every future term forever.
        if term > self.current_term:
            self._step_down(term)
            term_changed = True

        # 3. Grant the vote if we haven't voted in this term yet
        #    (or this is a repeat request from the same candidate)
        #    and the candidate's log is not behind ours.
        granted = False
        if (self.voted_for in (None, candidate_id)
                and self._log_is_up_to_date(last_log_index, last_log_term)):
            self.voted_for = candidate_id
            granted = True
            self.reset_election_timer()

        # 4. Persist before replying over the network if anything changed.
        if term_changed or granted:
            self._persist()

        return (self.current_term, granted)

Zab: A protocol written for a single product

Zab (ZooKeeper Atomic Broadcast) is a consensus protocol developed specifically for the Apache ZooKeeper system. Zab is designed to provide atomic (indivisible) message broadcasting with total ordering and fault tolerance. This means that all ZooKeeper nodes apply all transactions (state changes) in the same order, resulting in identical states. Zab is similar in purpose to Paxos, but is designed as a highly specialized master-replication (primary-backup) protocol with an emphasis on ease of implementation in a production product. It predates Raft by several years, solving the same problem with Multi-Paxos independently. It uses a model with a single leader (primary) and a set of followers—the same design later independently developed by Raft. Zab operates in four phases, and it's worth distinguishing them, as they are often conflated in descriptions. Leader Election: nodes agree on who will be the leader. Discovery: the elected leader collects the quorum's most recent epochs, calculates a new one, and ensures that the quorum accepts it. Synchronization: the leader aligns the followers' logs with its own: it sends missing data and forces them to roll back unnecessary data. Only then does Broadcast: client requests are accepted. The first three phases are collectively called recovery. In the ZooKeeper implementation, Discovery and Synchronization are partially merged, but logically they are distinct steps: the former establishes an epoch, the latter aligns the history. As long as the cluster has a leader and a quorum of living nodes, all client operations sequentially pass through the leader and are distributed to followers.

Recovery: Elections, Epoch, Synchronization

The process of starting or restoring a cluster begins with an election. ZooKeeper uses FastLeaderElection for this: nodes exchange votes and compare candidates using the tuple (epoch, zxid, sid)—epoch first, then the latest zxid if there's a tie, and only if the history is completely identical does the node with the higher ID win. Ultimately, the node with the most complete history in the quorum becomes the leader, and the sid is needed only to resolve ties deterministically. The strict maximum requirement distinguishes Zab from Raft, where a "no worse" log is sufficient—we'll return to this below. After the election, one node becomes the leader, and the others become its followers.

Discovery: Epoch Establishment. The leader doesn't immediately begin accepting operations—it must first secure a new epoch. It collects the followers' last known epochs (the CEPOCH message), takes the maximum, increments it by one, and broadcasts the result back (NEWEPOCH). A follower who accepts a new epoch thereby commits not to accept anything from leaders of previous epochs. Once a new epoch has been accepted by a quorum, the previous leader, even if still alive and unaware of their resignation, can no longer commit, their proposals will be rejected as obsolete.

This is exactly the same work performed by the Prepare phase in Paxos. The difference is in frequency: Paxos runs it for every value (except for optimized Multi-Paxos), while Zab runs it once per leader change.

Synchronization: Log Alignment. Only now does the leader consolidate the quorum's history. Each node stores the latest zxid—a transaction identifier consisting of the epoch number and a counter within it. The leader requests each follower's zxid and checks it against its own log:

If a follower is missing any transactions, the leader sends it the missing log records to bring it up to its latest transaction.

If a follower has "extra" records (with higher zxid values, possibly resulting from the previous leader not having time to commit these records), the leader instructs the follower to roll back these unconfirmed changes. This is necessary because, since they weren't previously committed by the majority, they shouldn't affect the system's state after recovery.

When the leader and follower reach the same latest zxid value, the leader includes this follower in the synced state. Having gathered a quorum of such followers, it broadcasts NEW_LEADER, a signal that the history is aligned and ready to proceed. From this point on, all followers in the quorum have the same history prefix, and the leader proceeds to the broadcast phase.

The order here is not arbitrary. First the epoch, then the logs—because rolling back someone else's records is a destructive operation, and permission to do so must be obtained in advance. If a leader began reconciling history before a quorum had recognized its epoch, it could delete transactions from followers and then be abandoned—and the data already reported to the client would disappear. Recognizing the epoch by a quorum ensures that subsequent events won't be undone by an older leader.

Atomic broadcast

In the broadcast phase, the current leader accepts client requests (transactions, such as changes to ZK nodes) and distributes them to all followers. Each new request receives a unique incrementing zxid (consisting of the leader's epoch number and a counter). The leader sends a proposal with this zxid to all followers (essentially a record of the transaction). Followers, having received a proposal, record it in their log and send an acknowledgment (ACK) to the leader. When the leader collects ACKs from a quorum of nodes, it sends a commit command to everyone for that zxid - from that moment on, the transaction is applied to everyone. Structurally, this is half of Paxos: proposal plays the role of an Accept request, and the Prepare phase is not needed, because the leader has already set its epoch during the recovery phase - just like Multi-Paxos skips Prepare until the leader changes. There is no separate message about a commit in basic Paxos, notification to learners is not considered part of the protocol, in Zab this is an explicit step. In ZooKeeper, all changes go through this negotiation, which ensures a linear order: if one transaction A was sent before B and confirmed, then all nodes will apply A before B. The leader takes care of this by ordering outgoing proposals, and the overall confirmation queue is maintained by a quorum requirement.

Zab guarantees reliable delivery (every transaction confirmed by the leader will reach all correct nodes) and total order of delivery. In addition, the causal order is respected: if transaction B is sent after the delivery of A on the same originator node, then B will be ordered after A. These guarantees correspond to the semantics of ZooKeeper: all nodes see changes in the same order, and a client sequentially sending requests sees them in the order they were sent.

Why ZooKeeper didn't take Paxos

The obvious question is: why write your own protocol when Multi-Paxos solves the same problem? The answer is that ZooKeeper needs a guarantee that Paxos doesn't provide.

Paxos ensures consensus on each log slot independently. Slots can be filled in any order, and one leader is free to begin work on slot 5 without waiting for the outcome of slot 4. For a database that issues commands when the entire prefix is ready, this is fine. For ZooKeeper, it's not: its transactions are incremental and written relative to state. Writing "increment znode /config version from 7 to 8" is only meaningful if the state is actually at version 7. Swap two such transactions or lose one in the middle, and the result won't be "slightly different," it'll be meaningless.

Therefore, Zab requires primary order—two conditions beyond the usual total order. First, a leader's proposals are delivered in the order in which they issued them, with no gaps in the middle. Second, if the leader of epoch e could have seen a transaction—that is, it was proposed in an epoch earlier than e and is potentially visible—then it is ordered before anything the leader proposes. To meet this second requirement, the Discovery phase elects the node with the highest zxid as the leader and forces the other nodes to roll back any tails the leader doesn't have.

This explains the difference in the election rules. Raft votes for a candidate whose log is "no worse" than the voter's log, and allows for the less advanced node to win, as long as it contains all committed records, it will simply overwrite any missing ones later. Zab requires the leader to have the highest zxid among the quorum and moves the alignment to a separate phase before work begins. Both approaches are safe, but they distribute work differently: Raft switches to servicing clients faster and aligns logs on the fly, while Zab does this in advance.

Fault Tolerance and Recovery in Zab

The recovery mechanics are discussed above, but the practical result is this: a transaction caught by the leader's failure is either recovered if it has reached the majority, or discarded. The monotonically increasing epoch number within zxid plays the same role as a term in Raft and a proposal number in Paxos: it prevents the previous leader from interfering with the new leader.

Zab was written for a specific implementation, and this is evident in the details. The leader writes the transaction to disk before sending a proposal, not after—otherwise, it could confirm to the client something it wouldn't survive. Proposals and ACKs are batched: under high load, a single fsync can process dozens of transactions, and throughput in practice is determined by this batching. The flip side of that specialization is its narrowness. Zab assumes a single leader and a full replica on each node, and is rarely encountered outside of coordination services: it's a single-product protocol, not a reusable component.

Comparison table of algorithms

The three protocols solve the same problem and differ in the payment method. Let's summarize the differences discussed above, and then we'll discuss where each is applied in practice.

	Paxos (Multi-Paxos)	Raft	Zab
Leader election	No mechanism, leadership is de facto, left to the implementation	Term-based voting, randomized timeouts	FastLeaderElection, a separate phase
Leader requirement	Any proposer that completed Prepare on a quorum	Log no less up-to-date than the voter's	Strictly highest zxid in the quorum
Who reconciles logs	Leader reopens unresolved slots as needed	Leader overwrites divergence on the fly, while serving	A separate phase before serving begins
Rounds per write	1 (Prepare is skipped while the leader is stable)	1 (AppendEntries + ACK)	1 (proposal + ACK), plus an explicit commit
Log ordering	Slots are independent, gaps possible	Strict prefix, no gaps	Strict prefix + primary order
Entries from a previous epoch	Reopened under a new number	Not committed by replica count — only indirectly, via committing an entry of the current term	Recovered or discarded during Synchronization
Membership changes	Not specified	Joint consensus or one-node-at-a-time	Dynamic reconfiguration (ZK 3.5+)
Main carriers	Spanner, Chubby, Ceph Monitors	etcd, Consul, CockroachDB, TiKV, KRaft	ZooKeeper only

Applications of Paxos, Raft, and Zab in real-world systems

Below is where these protocols run in practice, organized by algorithm. The tasks themselves are almost always the same: cluster configuration storage, a coordinator with master election and locks, and data replication within the DBMS. The only difference is which protocol was available when the system was written.

Paxos in Google services and distributed databases

Paxos has become the foundation for a number of Google's internal systems. In particular, the globally distributed Google Spanner database uses Multi-Paxos for synchronous replication between data centers. Each data fragment (split) is stored with multiple replicas, which form a separate Paxos group with its own leader. If the leader fails, the group elects a new one, and the service remains available. However, external consistency (i.e., strict serializability with respect to the actual order of transactions globally) does not follow from Paxos alone. Paxos orders operations within a single replica group, while a transaction in Spanner can affect multiple groups—they are coordinated by a two-phase commit on top of Paxos groups. Ordering between transactions, which have no data overlap and are executed in different regions, is provided by TrueTime—a time API with clearly bounded error, built on atomic clocks and GPS receivers in data centers. TrueTime returns not a moment, but an [earliest, latest] interval, guaranteed to contain real time. This is the basis of the commit-wait technique: a transaction, having received a commit timestamp, waits until the uncertainty interval has elapsed and only then is it considered committed. This ensures that a transaction started after another transaction completes will receive a strictly higher timestamp. Without TrueTime, Spanner would be just Paxos replication, and its use as an example of global consistency would be meaningless.

Another well-known example is Chubby, Google's distributed locking service, which also implements consensus based on Paxos. Chubby stores small but critical configuration data (such as master node coordinates) and uses Paxos for high availability: even if nodes fail, consistent storage remains for clients. Paxos is also used for metadata replication in distributed file systems, though more often indirectly: GFS relied on Chubby, i.e., Paxos via an external locking service, rather than implementing it itself. HDFS took a different approach and uses its own quorum mechanism for the edit log—Quorum Journal Manager. While not technically Paxos, its recovery procedure is based on the same idea: a new writer first secures an epoch number from a quorum of JournalNodes and only then reconciles the history. The difference lies more in the scope of the task: QJM doesn't agree on arbitrary values, but on a sequence of log segments with a single active writer. Among file systems, Ceph uses Paxos directly — in its monitor service (Ceph Monitors), to agree on the cluster map.

This is a special case where consensus isn't the foundation of the system, but a one-time operation. Cassandra is built on eventual consistency and has no leader, but for conditional writes ("insert only if the key doesn't already exist") it requires true consensus: two clients mustn't simultaneously agree that the key doesn't exist. For each such operation, Lightweight Transactions performs a full Paxos round trip over the replicas of that key—four message rounds versus one for a regular write. The cost is so significant that LWT is used selectively rather than as the default mode. In 4.1, they attempted to mitigate this with a rewritten Paxos implementation (v2), but the nature of the overhead remained unchanged. A radical solution is Accord, a protocol without a dedicated leader that delivers a transaction in a single round when transactions don't conflict. It was included in the 6.0 branch, which currently (August 2026) is in alpha status, it is not in the stable 5.0 branch.

Cassandra shows a limit here: consensus does not have to be the architecture of the entire system, it can be a local tool for a single operation – and in this form its cost is most clearly visible.

Raft in orchestration systems, service discovery, and new DBMSs

Over the past ten years, Raft has displaced Paxos from almost all new open-source projects requiring consensus.

The most notable example is etcd, where Raft replicates the change log between nodes. The entire Kubernetes cluster state passes through etcd, so each container deployment is essentially a Raft commit. Consul uses Raft only for the portion of data that requires consistency: the service catalog, configuration keys, sessions, and locks. Everything else lives outside of consensus, a distinction we'll return to below.

In distributed DBMSs, Raft has become the basis for replication at the level of individual data ranges. This is because a single Raft group doesn't scale for writes: all commands pass through a single leader, and its disk and network interface card become the limiting factor for the entire cluster. The solution isn't just one Raft per system, but many: CockroachDB and YugabyteDB maintain a separate group for each range, with leaders of different groups residing on different nodes, and the load is spread across the cluster. Each group provides linearizability within its range and automatically switches leaders in the event of a node failure, transactions affecting multiple ranges are coordinated separately across groups. TiKV (the storage engine behind TiDB) is structured similarly, and its Raft implementation is extracted into a standalone Rust library, raft-rs, which is also reused by third-party projects.

Zab and ZooKeeper in coordination systems

Zab is used almost exclusively within ZooKeeper—it's a protocol written specifically for the product. ZooKeeper itself, however, has been the foundation of the Hadoop and Apache ecosystems for decades: distributed locks, master election, configuration nodes, and queues were built on it. For over a decade, it served as the coordinator for Apache Kafka: storing broker information, topic and partition configuration, ensuring cluster controller election, and ensuring that exactly one broker is the leader of a given partition. This dependency has now been eliminated. As part of KIP-500, Kafka received its own built-in consensus, KRaft, where dedicated controller nodes form a quorum and store metadata in the internal __cluster_metadata log. KRaft was declared production-ready in version 3.3 (2022), ZooKeeper mode was deprecated in 3.5, and 3.9 was the last version with its support. In Kafka 4.0, released on March 18, 2025, ZooKeeper support was completely removed, leaving KRaft as the only operating mode. A direct upgrade from a ZooKeeper-based cluster to 4.0 is not possible: an intermediate migration to KRaft in a 3.x version is required. However, a large number of 3.x installations still use ZooKeeper, so the Kafka + ZK combination will be around for a long time to come.

Another example is HDFS NameNode HA: ZooKeeper is used for automatic failover between Active and Standby NameNodes. This is accomplished through a separate ZKFailoverController (ZKFC) process next to each NameNode: it holds an ephemeral znode lock in ZK, and the disappearance of the failed node's session triggers failover. The metadata itself is not stored in ZooKeeper, the edit log is replicated via the Quorum Journal Manager, the same epoch mechanism discussed above. HA shouldn't be confused with HDFS Federation: the latter solves a completely different problem—horizontal scaling through multiple independent namespaces—and doesn't require ZooKeeper. In HBase (a NoSQL database on Hadoop), ZooKeeper stores information about the active region servers.

Where does the algorithm end?

Knowing Raft doesn't mean understanding etcd. The protocol defines how to maintain log consistency across nodes. Everything the system user sees—the data model, read guarantees, quorum loss behavior, and database growth limits—lies outside the protocol and is addressed by the implementation. Below are a few questions the protocol doesn't answer, and etcd, ZooKeeper, and Consul answer differently.

Who constitutes a quorum?

The protocol says "a majority of nodes," but doesn't specify which nodes actually count.

In etcd and ZooKeeper, all cluster members vote, and this limits cluster size: three or five nodes, going beyond that doesn't pay off. Consul is designed differently. A Raft quorum is formed by three to five servers, and each application machine runs a client agent that doesn't participate in consensus: it registers local services, performs health checks, caches responses, and proxies requests to servers. A cluster of three machines and a cluster of three thousand are the same Raft.

ZooKeeper arrived at a similar solution in a different way. Observers receive updates from the leader but don't vote, and they can be deployed in remote data centers—reads scale, and the quorum doesn't slow down.

Neither Raft nor Zab have either: they're layers on top of the protocol, and they determine the scale of the system more than the choice of protocol itself.

This also applies to the data model. etcd v3 stores a flat, sorted keyspace: the familiar /registry/pods/default/nginx hierarchy is the naming convention, and "get a subtree" is expressed as a range query by prefix. ZooKeeper retains a true znode tree with ephemeral and sequential nodes, and distributed locks, queues, and barriers are built on these primitives. Consul maintains a separate service catalog with its own schema on top of KV. Three data models are used on two algorithms because consensus agrees on the command log, and the commands in that log are of no concern to it.

What the reader sees

Neither Raft nor Zab offer linearizable reads for free: a quorum commit doesn't mean the write is visible on all nodes simultaneously. Who is responsible for freshness and when is a product-specific decision, and this is where the three systems diverge most significantly.

etcd delivers freshness by default. Regular reads go through leadership confirmation, and the client is guaranteed to see the result of the last successful write. For those who prioritize speed, there's a serializable mode: the response is returned by a local replica, which may be out of date.

ZooKeeper doesn't guarantee freshness by default. Reads are served by any server from its local replica, so a client can see a state that lags behind the last committed write. The only guarantee is that it won't see history backwards: the order of operations performed by a single client is preserved, and a revision once seen cannot be rolled back. This is sequential consistency, not linearizability. The difference is that linearizability requires not only the overall order of operations but also consistency with real time: a confirmed write must be seen by any subsequent read, regardless of which node it comes from. ZooKeeper provides the former and not the latter. To ensure a fresh read, the client calls sync before reading. ZooKeeper's ability to scale reads across all nodes relies precisely on this relaxation: nodes respond locally because they don't need to be fresh.

Consul devolves the choice to the individual request level: consistent goes through a quorum with confirmation of leadership, stale allows any server from its replica to respond, and the default mode is intermediate: the leader responds immediately based on its lease, so in theory it can return stale data if it has already been displaced but doesn't yet know it. Agent caching is also available, but it is enabled by an explicit flag and isn't enabled by default.

The practical consequence is simple: code that reads a key immediately after someone else's write and expects to see it works in etcd and silently breaks in ZooKeeper.

What happens when quorum is lost?

The protocol only says "progress stops." It doesn't describe what the system does in this case.

etcd rejects writes, and in linearizable mode, reads are also rejected, as the node can't confirm leadership and responds with an error. ZooKeeper stops writes but continues to serve reads from local replicas: they weren't fresh anyway, so degradation is gradual. Consul stops writing to the directory, but reads in stale mode continue to be served by servers, and agents continue to perform health checks locally. The results of these checks won't be written to the directory, and applications will receive an increasingly outdated picture.

The difference here is in the form of the rejection. In the first case, the application will receive an error and learn about the problem immediately. In the third, it will receive a plausible but outdated response and learn about the problem later, sometimes much later.

What happens to the log in production?

Raft describes the mechanics of log compaction, but not the policy: when to take a snapshot, what to do with old data, where growth limits lie. This is left to the implementation, and this is where production most often breaks.

MVCC, which gives etcd a revision history and the ability to subscribe to changes from a given point, comes with its main gotcha: old revisions are not deleted automatically. They are removed by compaction, and the freed space is returned to the file system only by defragmentation—a separate operation that locks the node for the duration. Without both, the database grows, and when --quota-backend-bytes (2 GB by default) is reached, the cluster goes read-only with a NOSPACE alarm. For Kubernetes, this means the cluster continues to operate but stops accepting any changes—a state that must be exited manually.

The second source of growth in etcd is watch subscriptions: each controller monitoring keys maintains a thread, and in large clusters, there are thousands of them.

ZooKeeper writes snapshots and the transaction log to disk, but doesn't delete old files. This is controlled either by autopurge in the configuration or by the administrator. A forgotten setting is a common cause of disk filling on a ZK node.

What you'll have to configure manually

Not a single article about Raft specifies the proper election timeout. Too short, and the cluster re-elects the leader at every spike in network latency, too long, and downtime during a real failure extends to seconds. The starting point is usually an order of magnitude higher than the typical RTT between nodes, and then it's adjusted based on metrics.

Second, disk. Commit in Raft is limited by the leader's fsync, so SSD is essential: network storage with unpredictable latency turns the cluster into a generator of false re-elections.

Third, monitoring. A cluster of three nodes, where one has been down for a long time, appears healthy: there's a quorum, writes are going through. But there's no more headroom, and the next failure will stop writes. You need to monitor the number of alive nodes and the commit latency—the fact that "is it responding or not" says nothing about the safety margin.

What no one else has

None of the three algorithms are designed for a quorum that spans across regions: each write requires a network round-trip to reach a majority, and interregional latency translates directly into commit latency.

Therefore, etcd has no built-in geo-replication—the cluster is kept within a single data center or nearby zones. Consul for multi-DC sets up independent Raft quorums in each data center and loosely links them by exchanging service registrations, without global consensus. Global consistency in systems that require it is built on top of, rather than through, consensus: in Spanner, through TrueTime and a two-phase commit between Paxos groups, in Cosmos DB, through a separate interregional replication mechanism with a selectable consistency level.

When consensus isn't needed

Before setting up a cluster, it's worth checking whether it's needed at all.

Eureka, the Netflix service registry, operates without consensus at all: each node accepts records independently and replicates them to the other nodes for best-effort. When a network partition occurs, nodes on either side diverge but continue to respond. For a registry where a client still rechecks the instance's availability when accessing it, this is a reasonable tradeoff—an outdated list of addresses is more useful than an unavailable registry.

The same technique is used within Consul. Cluster membership and failure detection are based on the Serf gossip protocol, where a few seconds of desynchronization is acceptable and cheaper than quorum, while the service catalog is written using Raft. The line is drawn where divergence ceases to be harmless: a node's life can be detected with a delay, but lock ownership cannot.

Consensus is expensive, and mature systems use it selectively.

Conclusion

The practical choice today seems simple. For new code, choose Raft: there are more implementations, the ecosystem is more vibrant, and its clarity really pays off during debugging. ZooKeeper is used in two situations: you're part of a Hadoop ecosystem or serving those same Kafka 3.x installations discussed above, or you need its primitives on a znode tree: ephemeral nodes and watches provide locks and queues for next to nothing, but building them on top of a KV storage system is significantly more expensive. Paxos is rarely used in new open source projects. It lives inside large systems that were written before Raft and have since inherited it—Spanner, Chubby, Ceph. It's always a custom implementation within the product, not a ready-made component that can be plugged in.

But before choosing, it's worth asking the question up front: is consensus even necessary? It buys consistency at the cost of delays for each write and stalls when quorum is lost. If a discrepancy of a few seconds is harmless, gossip is cheaper. If operations are commutative, CRDTs can handle it without coordination. If the load fits on a single node, a replica with manual failover is more reliable than a three-machine cluster that no one knows how to fix.

Where consensus is truly needed, it's increasingly hidden inside something else: Kubernetes carries etcd, Kafka carries KRaft, CockroachDB carries a Raft group for each data range. Understanding how this works is essential for debugging: to understand why the cluster crashed and what went wrong.