Abstract
This article objective is to cover key design and deployment considerations for High Availability Apache Kafka service.
Introduction
Kafka’s through its distributed design allows a large number of permanent or ad-hoc consumers, being highly available and resilient to node failures, also supporting automatic recovery. These characteristics make Kafka an ideal fit for communication and integration between components of large scale data systems.
Kafka application resilience is not enough to have a true HA system. Consumers and producers or the network design need also to be HA. Zookeepers and brokers need to be able to communicate among themselves, producers and consumers need to be able to access Kafka API. A fast car can reach the maximum speed only on a good track. A swamp is a swamp for a fast or a slow car.
Kafka Design
Kafka utilizes Zookeeper for storing metadata information about the brokers, topics, and partitions. Writes to Zookeeper are only performed on changes to the membership of consumer groups or on changes to the Kafka cluster itself. This amount of traffic is minimal, and it does not justify the use of a dedicated Zookeeper ensemble for a single Kafka cluster. Many deployments will use a single Zookeeper ensemble for multiple Kafka clusters.
Prior to Apache Kafka 0.9.0.0, consumers, in addition to the brokers, utilized Zookeeper to directly store information about the composition of the consumer group, what topics it was consuming, and to periodically commit offsets for each partition being consumed (to enable failover between consumers in the group). With version 0.9.0.0, a new consumer interface was introduced which allows this to be managed directly with the Kafka brokers. However, there is a concern with consumers and Zookeeper under certain configurations. Consumers have a configurable choice to use either Zookeeper or Kafka brokers for committing offsets, and they can also configure the interval between commits. If the consumer uses Zookeeper for offsets, each consumer will perform a Zookeeper write at every interval for every partition it consumes. A reasonable interval for offset commits is 1 minute, as this is the period of time over which a consumer group will read duplicate messages in the case of a consumer failure. These commits can be a significant amount of Zookeeper traffic, especially in a cluster with many consumers, and will need to be taken into account. It may be necessary to use a longer commit interval if the Zookeeper ensemble is not able to handle the traffic. However, it is recommended that consumers using the latest Kafka libraries use Kafka brokers for committing offsets, removing the dependency on Zookeeper. Outside of using a single ensemble for multiple Kafka clusters, it is not recommended to share the ensemble with other applications, if it can be avoided.
Kafka is sensitive to Zookeeper latency and timeouts, and an interruption in communications with the ensemble will cause the brokers to behave unpredictably. This can easily cause multiple brokers to go offline at the same time, should they lose Zookeeper connections, which will result in offline partitions. It also puts stress on the cluster controller, which can show up as subtle errors long after the interruption has passed, such as when trying to perform a controlled shutdown of a broker. Other applications that can put stress on the Zookeeper ensemble, either through heavy usage or improper operations, should be segregated to their own ensemble.
Infrastructure and Network Design Challenges
Application is the last layer in top of other 6 layers of OSI stack, including Network, Data Link and Physical. A power source that is not redundant can take out the rack switch and none of the servers in the rack are accessible. There are at least two issues with that implementation, e.g. non-redundant power source for the switch, lack of redundancy for the actual rack switch. Add that there is a single communication path from consumers and producers and exactly that is down so your mission critical system does not deliver the service that is maybe your main stream of revenue. Add to that rack-awareness is not implemented and all topic partitions reside in the infrastructure hosted by that exact rack that failed due a bad switch or a bad power source.
Did it happen to you? What was the price that you afforded to pay when that bad day came?
Network Design and Deployment Considerations
Implementing a resilient Kafka cluster is similar with implementing a resilient HDFS cluster. Kafka or HDFS reliability is for data/app reliability, at most compensating for limited server failure, but not for network failure, especially in cases when infrastructure and network has many single points of failure coupled with a bad deployment of Kafka zookeepers, brokers or replicated topics.
Dual NIC, dedicated core switches and redundant, rack top switches, balancing replicas across racks are common for a good network design for HA. Kafka, like HDFS, supports rack awareness. A single point of failure should not impact more than one zookeeper or one broker, or one partition of a topic. Zookeepers and Brokers should have HA communication, if one path is down, another path is used. They should be distributed ideally in different racks.
Network redundancy needs to provide the alternate access path to brokers for producers and consumers, when failure arises. Also, as a good practice, brokers should be distributed across multiple racks.
Configure Network Correctly
Network configuration with Kafka is similar to other distributed systems, with a few caveats mentioned below. Infrastructure, whether on-premises or cloud, offers a variety of different IP and DNS options. Choose an option that keeps inter-broker network traffic on the private subnet and allows clients to connect to the brokers. Inter-broker and client communication use the same network interface and port.
When a broker is started, it registers its hostname with ZooKeeper. The producer since Kafka 0.8.1 and the consumer since 0.9.0 are configured with a bootstrapped (or “starting”) list of Kafka brokers. Prior versions were configured with ZooKeeper. In both cases, the client makes a request (either to a broker in the bootstrap list, or to ZooKeeper) to fetch all broker hostnames and begin interacting with the cluster.
Depending on how the operating system’s hostname and network are configured, brokers on server instances may register hostnames with ZooKeeper that aren’t reachable by clients. The purpose of advertised.listeners is to address exactly this problem; the configured protocol, hostname, and port in advertised.listeners is registered with ZooKeeper instead of the operating system’s hostname.
In a multi-datacenter architecture, careful consideration has to be made for MirrorMaker. Under the covers, MirrorMaker is simply a consumer and a producer joined together. If MirrorMaker is configured to consume from a static ID, the single broker tied to the static IP will be reachable, but the other brokers in the source cluster won’t be. MirrorMaker needs access to all brokers in the source and destination data center, which in most cases is best implemented with a VPN between data centers.
Client service discovery can be implemented in a number of different ways. One option is to use HAProxy on each client machine, proxying localhost requests to an available broker. Synapse works well for this. Another option is to use a Load Balancer appliance. In this configuration, ensure the Load Balancer is not public to the internet. Sessions and stickiness do not need to be configured because Kafka clients only make a request to the load balancer at startup. A health check can be a ping or a telnet.
Distribute Kafka brokers across multiple racks
Kafka was designed to run within a single data center. As such, we discourage distributing brokers in a single cluster across multiple data centers. However, we recommend “stretching” brokers in a single cluster across multiple racks. It could be in the same private network or over multiple private networks within the same data center. There are considerations that each enterprise is responsible for implementing resilient systems.
A multi-rack cluster offers stronger fault tolerance because a failed rack won’t cause Kafka downtime.
However, in this configuration, prior to Kafka 0.10, you must assign partition replicas manually to ensure that replicas for each partition are spread across availability zones. Replicas can be assigned manually either when a topic is created, or by using the kafka-reassign-partitions command line tool. Kafka 0.10 or later supports rack awareness, which makes spreading replicas across racks much easier to configure.
At the minimum, for the HA of your Kafka-based service:
- Use dedicated “Top of Rack” (TOR) switches (can be shared with Hadoop cluster).
- Use dedicated core switching blades or switches.
- If deployed to a physical environment, then make certain to place the cluster on in a VLAN.
- For the switch communicating between the racks you will want to establish HA connections.
- Implement Kafka rack-awareness configuration. It does not apply retroactively to the existent zookeepers, brokers or topics.
- Test periodically your infrastructure for resiliency and impact on Kafka service availability, e.g. disconnect one rack switch impacting a single broker for example. If a correct design of the network and topic replication was implemented, producers and consumers should be able to work as usual, the only acceptable impact should be due to reduced capacity producers may accumulate a lag in processing or consumers may have some performance impact, however, everything else should be 100% functional.
- Implement continuous monitoring of producers and consumers to detect failure events. Alerts and possibly automated corrective actions.
See below a simplified example of a logical architecture for HA.
The above diagram shows an implementation for HA for an enterprise that uses 5 racks, 5 zookeepers distributed across multiple racks, multiple brokers distributed across those 5 racks, replicated topics, HA producers and consumers. This is similar with HDFS HA network design.
Distribute ZooKeeper nodes across multiple racks
ZooKeeper should be distributed across multiple racks as well, to increase fault tolerance. To tolerate a rack failure, ZooKeeper must be running in at least three different racks. Obviously, there is also the private network concept of failure. An enterprise can have zookeepers separated in different networks. That comes at the price of latency. It is a conscious decision between performance and reliability. Separating in different racks is a good compromise. In a configuration where three ZooKeepers are running in two racks, if the rack with two ZooKeepers fails, ZooKeeper will not have quorum and will not be available.
Monitor broker performance and terminate poorly performing brokers
Kafka broker performance can decrease unexpectedly over time for unknown reasons. It is a good practice to terminate and replace a broker if, for example, the 99 percentile of produce/fetch request latency is higher than is tolerable for your application.
Datacenter Layout
For development systems, the physical location of the Kafka brokers within a datacenter is not as much of a concern, as there is not as severe an impact if the cluster is partially or completely unavailable for short periods of time.
When serving production traffic, however, downtime means dollars lost, whether through loss of services to users or loss of telemetry on what the users are doing. This is when it becomes critical to configure replication within the Kafka cluster, which is also when it is important to consider the physical location of brokers in their racks in the datacenter.
If a deployment model across multiple racks and rack-awareness configuration was not implement prior to deploying Kafka, expensive maintenance to move servers around may be needed. The Kafka broker has no rack-awareness when assigning new partitions to brokers so everything to the date when the rack-awareness was implemented could be brittle to failure. This means that it cannot take into account that two brokers may be located in the same physical rack, or in the same network, therefore can easily assign all replicas for a partition to brokers that share the same power and network connections in the same rack. Should that rack have a failure, these partitions would be offline and inaccessible to clients. In addition, it can result in additional lost data on recovery due to an unclean leader election.
The best practice is to have each Kafka broker in a cluster installed in a different rack, or at the very least not share single points of failure for infrastructure services such as power and network. This typically means at least deploying the servers that will run brokers with dual power connections (to two different circuits) and dual network switches (with a bonded interface on the servers themselves to failover seamlessly). Even with dual connections, there is a benefit to having brokers in completely separate racks.
From time to time, it may be necessary to perform physical maintenance on a rack or cabinet that requires it to be offline (such as moving servers around, or rewiring power connections).
Other Good Practices
I am sure I missed other good practices, so here are a few more:
- Produce to a replicated topic.
- Consume from a replicated topic (consumers must be in the same consumer group).
- Each partition gets assigned a consumer; need to have more partitions than consumers.
- Resilient producers – spill data to local disk until Kafka is available.
Other Useful References
Mirroring data between clusters: http://kafka.apache.org/documentation.html#basic_ops_mirror_maker
Data centers: http://kafka.apache.org/documentation.html#datacenters
Thanks
I'd like to thank all Apache Kafka contributors and the community that drives the innovation. I mean to thank everyone that contributes to improve documentation and publications on how to design and deploy Kafka as a Service in HA mode.
