Why a Global Cache Needs to Travel

Netflix's service is built on hundreds of microservices, most of them stateless and scaled independently. That statelessness only works if the state those services need — a member's viewing history, ratings, personalized recommendations — is available in whatever region happens to be serving the request. Traffic shifts between Netflix's AWS regions (Northern Virginia, Oregon, and Ireland) for many reasons: infrastructure problems, regional failovers, or deliberate Chaos Kong exercises. When that happens, a "cold" cache in the destination region would turn every miss into a slow database read. The answer is to replicate cached data between regions before it's needed there.

EVCache, Netflix's memcached-based RAM store for the cloud, has been in production for more than five years and handles the bulk of this caching. At peak, production deployments process upwards of 30 million requests per second across tens of thousands of memcached instances, storing hundreds of billions of objects — nearly 2 trillion requests per day globally. The challenge was extending that scale across regional boundaries.

There's a second use case that makes replication essential. Some caches don't front a database at all; they "memoize" data that's expensive to recompute. When a compute system writes that data to a local cache, there's no persistent store to fall back on. The only way for another region to serve that data is to replicate the cache itself.

Consistency Can Be Eventual

The replication design starts with what it explicitly does not require: strong global consistency. It's acceptable for Ireland and Virginia to briefly disagree on a recommendation, as long as it doesn't hurt browsing or streaming. That eventual-consistency model lets EVCache skip global locking, quorum reads and writes, transactional updates, and partial-commit rollbacks — all of which would add complexity and latency.

Two other constraints shaped the architecture. First, replication must never affect the performance or reliability of local cache operations, even when cross-region links degrade. Second, replication latency only needs to be "good enough" — fast enough that the occasional inconsistency isn't noticeable, but nothing approaching instantaneous. These are loose requirements, deliberately chosen to keep the system tractable.

How a Set Travels Between Regions

For a SET operation, the replication path is invisible to the calling application:

  1. The EVCache client library sends the SET to the local region's cache.
  2. The client also writes metadata — the key, but not the data — to a Kafka replication queue.
  3. A Replication Relay service in the local region reads messages from that queue.
  4. The Relay fetches the actual data for the key from the local cache.
  5. The Relay sends a SET request to the destination region's Replication Proxy service.
  6. The Replication Proxy performs a SET to its local cache and returns a success response.
  7. Local applications in the receiving region now see the updated value on GET.

The flows for DELETE and TOUCH follow the same pattern, minus the step of fetching the existing value. Only one message crosses region boundaries: the request from the Replication Relay to the Replication Proxy. Clients of EVCache are unaware other regions exist; all reads and writes go to local instances.

Components and Their Jobs

Three components cooperate to move data between regions. Each one runs independently of the application services and cache instances they serve.

Replication Message Queue

Kafka is the cornerstone. For a fully replicated cache, the Kafka stream has two consumers: one Replication Relay cluster per destination region. Dedicated clusters for each target decouple the replication paths, so latency or failure in one direction doesn't affect the other.

If a destination region becomes severely latent or fails outright for an extended period, the Kafka buffer fills and older messages are dropped. Those messages are simply never sent to the broken region. Services using replicated caches are designed to tolerate exactly this kind of disruption.

Replication Relay

The Relay cluster consumes messages from Kafka, then writes replication requests over a secure connection to the Replication Proxy cluster in the destination region, fetching the data from the local cache first when needed. It retries requests that time out or fail. If cross-region latency spikes, Kafka keeps accepting messages and buffers the backlog until the Relay catches up — the local cache is never blocked.

Replication Proxy

The Proxy cluster runs in the receiving region and synchronously writes incoming data to its local cache, then responds to the Relay so it knows the write succeeded. The Proxy uses the same open-source EVCache client any application would use, inheriting all the sharding, instance selection, retrying, and intra-region replication logic for free. Both the Relay and Proxy clusters run multiple instances across Availability Zones, letting them handle high traffic while surviving localized failures.

Metadata Only, Please

Two design choices keep the system efficient. First, the Kafka queue carries only key and metadata — never the cached payload. This keeps the Kafka deployment small and fast; holding all cache data in Kafka would make it a storage and network bottleneck. Instead, the payload is fetched from the local cache after the Relay reads the message, so no second copy is ever written to Kafka.

Second, relying on metadata alone means some caches don't need full data replication at all. For caches where a SET in one region only needs to invalidate the key elsewhere, the system sends a DELETE instead of the new value. A subsequent GET in the destination region misses and the application handles it like any other miss. When cross-region read traffic is low, occasional misses are cheaper than continuously shipping the data.

The result is an asynchronous, eventually consistent pipeline that tolerates regional outages, scales independently of local cache traffic, and never forces an application in one region to wait on the network conditions of another.

Tuning for Latency and Throughput

Every cache has different latency and throughput requirements, so we tune each one accordingly. For most of our caches, the 99th percentile of end-to-end replication latency is under one second. A portion of that time is an intentional buffer: we batch messages throughout the replication flow to boost throughput at the cost of a little latency. Our highest-volume replicated cache has a 99th percentile latency of only about 400ms because its buffers fill and flush so quickly.

Persistent connections between the Relay and Proxy clusters proved to be a major win. They eliminate the TCP 3-way handshake and the extra network round-trips needed for TLS/SSL session establishment before each replication request. Latency dropped significantly and became more stable.

We also batch multiple messages into a single request to fill the TCP window more efficiently, which improves throughput and lowers overall communication latency between the clusters. Ideally, the batch size would dynamically match the TCP window size as it fluctuates over a connection's lifetime; in practice, we tune it empirically for good throughput. Batching adds a small amount of latency, but it lets us extract more work from each TCP packet, reduces the number of connections each instance must maintain, and ultimately allows us to run fewer instances for a given replication demand profile. With these optimizations, EVCache's cross-region replication routinely handles over a million requests per second at daily peak.

Operational Lessons

Our Kafka-based replication system has been in production for over a year, replicating more than 1.5 million messages per second at peak. That journey hasn’t been without growing pains. We’ve seen periods of elevated end-to-end latencies with causes ranging from obvious (autoscaling rule problems in the Proxy application) to opaque (congestion on the cross-region link over the public Internet).

Before we moved into Amazon VPC, one of our biggest headaches was implicit packets-per-second limits on AWS instances. Hitting that cap triggered a cascade of TCP timeouts, dropped packets, high replication latencies, TCP retries, and failed replication requests that later had to be retried. The fix is straightforward: scale out. More instances mean more aggregate packets-per-second capacity, and sometimes two large instances are a better choice than a single extra large instance even when the cost is the same. Moving into VPC raised those packet-rate limits and gave us access to enhanced networking features, letting the Relay and Proxy clusters do more work per instance.

Diagnosing latency issues required visibility into each link of the chain. We introduced metrics to track latency at every stage: from the client application to Kafka, from the Relay cluster's reads of Kafka, from the Relay cluster to the remote Proxy cluster, and from the Proxy cluster to its local cache servers. End-to-end timing metrics monitor the overall health of the system.

Known Rough Edges

Several issues remain on our plate. Kafka isn't easy to scale on demand. When a cache needs more replication-queue capacity, we must manually add partitions, configure consumers with matching thread counts, and scale the Relay cluster accordingly. This manual process can lead to duplicate or re-sent messages, causing inefficiency and more eventual-consistency skew than usual.

Losing an EVCache instance in the remote region spikes latency while the Proxy cluster attempts and fails to write to the missing instance. That latency propagates back to the Relay side, which waits for confirmation on each batched request. We've reduced the duration of this state by detecting lost instances earlier and are investigating reconciliation mechanisms to soften the impact. Client-side changes also help Proxy instances cope when cache instances disappear.

Kafka monitoring for missing messages isn't an exact science. Software bugs can cause messages to never appear in a Kafka partition or to go unreceived by the Relay cluster. Our current approach compares the total messages received by the Kafka brokers (per topic) against the number the Relay cluster replicates, investigating whenever the difference exceeds a small acceptable threshold for any meaningful period. We also watch maximum latencies rather than averages, because a single slow partition warrants investigation even when the mean looks healthy. We continue refining these alerts to catch real problems with fewer false positives.

What's Next

The replication system still has room to grow. Potential improvements include pipelining replication messages over a single connection for better connection utilization, optimizing around the network's TCP window size, or migrating to the new Kafka 0.9 API. We'd also like the Relay clusters to autoscale cleanly without inflating latencies or increasing duplicate message rates.

EVCache delivers globally replicated data at RAM speed so any member request can be served from anywhere. Building reliable, fast replication for caching systems at global scale is an ongoing challenge, and we expect the design to keep evolving alongside our member base and our needs.