Kafka Inside Keystone Pipeline

Running Kafka at Netflix Scale

Netflix's Keystone pipeline is a unified event publishing, collection, and routing infrastructure that supports both batch and stream processing. At the heart of this system sit two sets of Kafka clusters: Fronting Kafka clusters that ingest messages from producers across virtually every application instance, and Consumer Kafka clusters that host a subset of topics routed by Samza for real-time consumers.

The operational footprint is substantial: 36 Kafka clusters comprising more than 4,000 broker instances, ingesting over 700 billion messages daily. The current transition is from Kafka version 0.8.2.1 to 0.9.0.1.

Availability Trade-offs

Given Kafka's architecture and the sheer data volume, achieving lossless delivery in AWS EC2 would be cost prohibitive. Netflix has instead worked with dependent teams to establish an acceptable data loss threshold while balancing cost, achieving a daily loss rate below 0.01%. Dropped messages are tracked so teams can respond when necessary.

The Keystone pipeline produces messages asynchronously to avoid blocking applications. If a message cannot be delivered after retries, the producer drops it to preserve application availability and user experience. This drives the production configuration:

• acks = 1
• block.on.buffer.full = false
• unclean.leader.election.enable = true

Most Netflix applications use a Java client library to produce to the pipeline. Each application instance runs multiple Kafka producers, with each one targeting a Fronting Kafka cluster for sink-level isolation. Producers support flexible topic routing and sink configuration driven via dynamic configuration, allowing runtime changes like traffic redirection or topic migration across clusters without restarts. Non-Java applications can instead send events to Keystone REST endpoints, which relay messages to the fronting clusters.

Producers intentionally avoid keyed messages for greater flexibility. Approximate ordering is re-established later in the batch processing layer (Hive/Elasticsearch) or in the routing layer for streaming consumers. Stability of the Fronting Kafka clusters is a top priority since they are the injection gateway; client applications are not permitted to consume directly from them, ensuring predictable load.

Cloud-Specific Challenges

Kafka was designed for data centers, but Netflix has made notable adaptations for the cloud. Instances have unpredictable lifecycles and can terminate at any time due to hardware issues. Transient network problems are expected. These are manageable for stateless services but create real challenges for stateful systems requiring ZooKeeper and a single controller.

Most incidents trace back to outlier brokers. An outlier might result from uneven workload, hardware problems, or environmental factors like noisy neighbors in multi-tenant infrastructure. Outliers respond slowly or experience frequent TCP timeouts and retransmissions. Producers sending to such brokers are likely to exhaust their local buffers waiting for responses, after which message drops become certain. The Kafka 0.8.2 producer compounds this by lacking timeout support for messages sitting in the buffer.

Replication improves availability but also creates inter-dependencies among brokers, allowing an outlier to cascade. Slow replication leads to lag, eventually forcing partition leaders to read from disk to serve replication requests. This slows affected brokers further and ultimately causes producers to drop messages due to exhausted buffers.

Early on, Netflix experienced an incident where a ZooKeeper issue caused significant message drops across a cluster with hundreds of instances, with little recourse for debugging in a short window. That experience drove efforts to reduce statefulness, detect outliers faster, and enable quick clean-state restarts when incidents occur.

Deployment Strategy

The operational approach favors many small clusters over one giant one, reducing per-cluster complexity. The largest cluster stays under 200 brokers, and each cluster maintains fewer than 10,000 partitions to improve availability and keep request latency manageable. Replicas for each topic are distributed evenly to simplify capacity planning and outlier detection. Each Kafka cluster uses its own dedicated ZooKeeper cluster to limit the blast radius of ZooKeeper issues.

Failover Automation

Netflix automated a failover process that diverts both producer and consumer traffic to a fresh Kafka cluster when the primary is unhealthy. Each fronting cluster has a cold standby with the desired launch configuration but minimal initial capacity. The standby starts with no topics and no shared ZooKeeper, guaranteeing a clean state. It also uses replication factor 1, eliminating any replication issues from the original cluster.

When failover triggers, the sequence is:

1. Resize the failover cluster to the desired size.
2. Create topics on and launch routing jobs for the failover cluster in parallel.
3. Optionally wait for partition leaders to be established by the controller, minimizing initial message drops.
4. Dynamically change producer configuration to switch traffic to the failover cluster.

Full automation keeps failover time under five minutes. Afterward, the original cluster can be debugged using logs and metrics, or destroyed and rebuilt with a new image before traffic is switched back. This same strategy is used for offline maintenance, including Kafka version upgrades without rolling upgrades or inter-broker protocol version configuration.

Custom Tooling

Producer sticky partitioner

A custom partitioner in the Java producer library sticks to one partition for a configurable duration before randomly choosing the next. Combined with linger, this improves message batching and reduces broker load.

Rack aware replica assignment

All Kafka clusters span three AWS availability zones, treating each zone as a rack. Custom replica assignment ensures replicas of a topic land in different zones. This protects against a full zone outage and also improves tolerance when multiple brokers on the same physical host are terminated together, exceeding Kafka's default N − 1 fault tolerance for replication factor N. This work was contributed upstream as KIP-36 and Apache Kafka Pull Request #132.

Kafka Metadata Visualizer

Kafka metadata lives in ZooKeeper, but Exhibitor's tree view is cumbersome for correlating information. Netflix built a custom UI with chart and tabular views, using color to indicate ISR state. Key capabilities include separate tabs for brokers, topics, and clusters; sortable and searchable data; cross-cluster topic search; direct broker ID-to-AWS instance mapping; and leader-follower relationship views.

Monitoring

A dedicated Kafka monitoring service tracks broker status, specifically whether a broker is offline from ZooKeeper. It also verifies each broker's ability to receive from producers and deliver to consumers by acting as its own producer and consumer for continuous heartbeat messages, measuring latency. For legacy ZooKeeper-based consumers, it checks partition counts per consumer group to ensure full consumption. For Keystone Samza routers, it compares checkpointed offsets against broker log offsets to detect stalled consumers or significant lag.

Beyond this, extensive dashboards monitor traffic flow down to individual topics along with most broker-level metrics.

Roadmap for the Keystone Pipeline

The team is mid-migration to Kafka 0.9, which brings several capabilities they intend to adopt: the new consumer APIs, producer message timeouts, and quotas. In parallel, the Kafka clusters are being moved into AWS VPC. The expectation is that VPC's improved networking compared to EC2 Classic will help raise availability and resource utilization.

A tiered SLA model for topics is also in the works. Topics that tolerate minor loss may be configured with just a single replica. Dropping replication saves bandwidth and reduces the state changes that depend on the controller, a step toward making Kafka less stateful in an infrastructure that favors stateless services. The obvious trade-off is potential message loss when a broker fails, but the producer message timeout in 0.9 and the use of AWS EBS volumes should help contain the damage.

Further posts from the Real-Time Data Infrastructure Team will cover routing infrastructure, container management, and stream processing.