Hello everyone. In this article, I will explain why we at Avito chose Kafka nine months ago and what it entails. I will share one of the use cases — a message broker. Finally, we will discuss the benefits we've gained from adopting a Kafka as a Service approach.

The Problem

First, a bit of context. Some time ago, we began transitioning away from a monolithic architecture, and now Avito has several hundred different services. Each service has its own storage, technology stack, and is responsible for its part of the business logic.

One of the challenges with a large number of services is communication. Service A often wants to know information that service B has. In this case, service A requests information from service B through a synchronous API. Service C wants to know what’s happening with services G and D, and those, in turn, are interested in services A and B. When there are many 'inquisitive' services, the connections become a tangled mess.

Moreover, at any moment, service A may become unavailable. What should service B and all other services relying on it do in that case? If a business operation requires a chain of sequential synchronous calls, the likelihood of the entire operation failing increases significantly (and this likelihood grows as the chain gets longer).

Technology Selection

Okay, the problems are clear. They can be resolved by creating a centralized messaging system between services. Now each service only needs to be aware of this messaging system. Additionally, the system itself must be fault-tolerant and horizontally scalable, and in case of failures, it should be able to buffer requests for later processing.

Let’s now choose the technology that will implement message delivery. To do this, we first need to understand what our expectations are:

• messages between services should not be lost;
• messages may be duplicated;
• messages can be stored and read for several days (persistent buffer);
• services can subscribe to data of interest;
• multiple services can read the same data;
• messages may contain detailed, large payloads (event-carried state transfer);
• sometimes a guarantee of message order is required.

It was critically important for us to choose a highly scalable and reliable system with high throughput (at least 100k messages at several kilobytes per second).

At this stage, we said goodbye to RabbitMQ (difficult to maintain stability at high rps), PGQ from SkyTools (not fast enough and poorly scalable), and NSQ (not persistent). We use all these technologies in our company, but they were not suitable for the task at hand.

Next, we began looking at new technologies for us — Apache Kafka, Apache Pulsar, and NATS Streaming.

We first eliminated Pulsar. We decided that Kafka and Pulsar are quite similar solutions. And despite the fact that Pulsar has been tested by large companies, is newer, and offers lower latency (theoretically), we chose Kafka as the de facto standard for such tasks. We may come back to Apache Pulsar in the future.

So we were left with two candidates: NATS Streaming and Apache Kafka. We studied both solutions in detail, and both fit the task. But ultimately, we were hesitant about the relative youth of NATS Streaming (and the fact that one of its main developers, Tyler Treat, decided to leave the project and start his own — Liftbridge). Moreover, the Clustering mode of NATS Streaming did not allow for strong horizontal scaling (this might no longer be an issue after the addition of the partitioning mode in 2017).

Nevertheless, NATS Streaming is a cool technology written in Go and has support from the Cloud Native Computing Foundation. Unlike Apache Kafka, it does not need Zookeeper to work (possibly, soon we will be able to say the same about Kafka), as it implements RAFT internally. Additionally, NATS Streaming is simpler to administer. We do not rule out the possibility of returning to this technology in the future.

Still, as of today, our winner is Apache Kafka. In our tests, it performed fast enough (over a million messages per second for both reading and writing with message sizes of 1 kilobyte), was reliable, well-scalable, and proven in production by large companies. Additionally, Kafka is supported by at least several major commercial companies (for instance, we use the Confluent version), and Kafka has a developed ecosystem.

Overview of Kafka

Before we start, I immediately recommend an excellent book — “Kafka: The Definitive Guide” (it is also available in Russian translation, but the terms can be a bit confusing). It contains information necessary for a basic understanding of Kafka and even a bit more. The documentation from Apache and the blog from Confluent are also well-written and easy to read.

So, let's take a bird's-eye view of how Kafka is structured. The basic topology of Kafka consists of a producer, consumer, broker, and zookeeper.

Broker

The broker is responsible for storing your data. All data is stored in binary format, and the broker knows little about what they represent and their structure.

Each logical type of event usually resides in its own separate topic. For example, a create event might go to the item.created topic, while an update event goes to item.changed. Topics can be seen as classifiers for events. At the topic level, configuration parameters can be set, such as:

• the amount of stored data and/or its age (retention.bytes, retention.ms);
• data redundancy factor (replication factor);
• maximum size of a single message (max.message.bytes);
• minimum number of in-sync replicas required to write data to the topic (min.insync.replicas);
• possibility of failing over to an out-of-sync lagging replica with potential data loss (unclean.leader.election.enable);
• and many others (https://kafka.apache.org/documentation/#topicconfigs).

In turn, each topic is divided into one or more partitions. Events ultimately end up in the partition. If there is more than one broker in the cluster, partitions will be evenly distributed among all brokers (as much as possible), allowing the load for writing and reading in one topic to be scaled across multiple brokers.

On disk, data for each partition is stored as segment files, which by default are equal to one gigabyte (controlled through log.segment.bytes). An important feature is that data removal from partitions (when retention is triggered) occurs in segments (it is not possible to remove a single event from a partition, only entire segments can be deleted, and only inactive ones).

Zookeeper

Zookeeper acts as a metadata store and coordinator. It can determine whether brokers are alive (this can be viewed from zookeeper's perspective using the zookeeper-shell command ls /brokers/ids), which broker is the controller (get /controller), are the partitions in sync with their replicas (get /brokers/topics/topic_name/partitions/partition_number/state). Additionally, the producer and consumer will first interact with zookeeper to determine which broker holds which topics and partitions. In cases where a replication factor greater than 1 is set for a topic, zookeeper will indicate which partitions are leaders (writes will occur in these, and reads will be from these as well). In the event of a broker failure, zookeeper will record the information about new leader partitions (as of version 1.1.0 asynchronously, and this is important).

In older Kafka versions, zookeeper was also responsible for storing offsets, but now they are held in a special topic __consumer_offsets on the broker (though you can still use zookeeper for these purposes).

The simplest way to turn your data into a pumpkin is really to lose information with zookeeper. In such a scenario, understanding what and where to read will be very difficult.

Producer

A Producer is most often a service that directly writes data to Apache Kafka. The producer selects a topic where its messages will be stored and starts writing information to it. For example, a producer could be an advertisement service. In this case, it will send events to the thematic topics such as 'advertisement created', 'advertisement updated', 'advertisement deleted', etc. Each event represents a key-value pair.

By default, all events are distributed among the topic's partitions round-robin if no key is set (losing order), and via MurmurHash (key) if a key is present (maintaining order within a single partition).

It should be noted here that Kafka guarantees the order of events only within a single partition. However, in practice, this often isn't an issue. For example, you can reliably add all changes for the same advertisement to one partition (thus preserving the order of these changes within the advertisement). You can also pass a sequence number in one of the event fields.

Consumer

The consumer is responsible for receiving data from Apache Kafka. Referring back to the previous example, a moderation service could act as the consumer. This service will subscribe to the ad service topic and will receive new ads as they appear, analyzing them for compliance with certain established policies.

Apache Kafka keeps track of the last events received by the consumer (this is done using a special topic __consumer__offsets), thereby ensuring that when a message is successfully read, the consumer does not receive the same message twice. However, if the enable.auto.commit = true option is used and the responsibility for tracking the consumer's position in the topic is entirely left to Kafka, it can result in data loss. In production code, the consumer's position is most often manually controlled (the developer manages when a commit of the read event must occur).

In cases where one consumer is not enough (for example, when the flow of new events is very large), several more consumers can be added by linking them together in a consumer group. The consumer group logically represents the same consumer, but with data distribution among the group members. This allows each member to take their share of messages, thus scaling the reading speed.

Test Results

I won't write a lot of explanatory text here; I'll simply share the results obtained. The testing was conducted on 3 physical machines (12 CPUs, 384GB RAM, 15k SAS DISK, 10GBit/s Net), with brokers and Zookeeper deployed in LXC.

Performance Testing

The following results were obtained during the testing.

• The write speed of 1KB messages simultaneously by 9 producers is 1,300,000 events per second.
• The read speed of 1KB messages simultaneously by 9 consumers is 1,500,000 events per second.

Fault Tolerance Testing

The following results were obtained during the testing (3 brokers, 3 Zookeepers).

• A non-standard shutdown of one of the brokers does not lead to the stoppage or inaccessibility of the cluster. Operation continues normally, but there is increased load on the remaining brokers.
• An unexpected termination of two brokers in a cluster of three brokers with min.isr = 2 results in the cluster being unavailable for writing, but still available for reading. If min.isr = 1, the cluster remains available for both reading and writing. However, this mode contradicts the requirement for high data integrity.
• The unexpected termination of one of the Zookeeper servers does not lead to the stopping or unavailability of the cluster. Operations continue as normal.
• An unexpected termination of two Zookeeper servers results in the cluster being unavailable until at least one Zookeeper server is operational again. This statement holds true for a Zookeeper cluster of 3 servers. As a result of the investigations, it was decided to expand the Zookeeper cluster to 5 servers to enhance fault tolerance.
  

Kafka as a service

We found that Kafka is an excellent technology that allows us to solve the task at hand (implementing a message broker). However, we decided to prevent services from directly accessing Kafka and placed a data-bus service on top of it. Why did we do this? There are actually several reasons.

• The data-bus took on all tasks related to integration with Kafka (implementing and configuring consumers and producers, monitoring, alerting, logging, scaling, etc.). Thus, integration with the message broker is as straightforward as possible.

• The data-bus allowed abstraction from a specific language or library for working with Kafka.

• The data-bus allowed other services to abstract away from the storage layer. Perhaps, at some point, we will switch from Kafka to Pulsar, and no one will notice (all services only interact with the data-bus API).

• The data-bus took on the validation of event schemas.

• Authentication has been implemented using the data-bus.

• Under the cover of the data-bus, we can update Kafka versions without downtime, invisibly, while centrally managing producer, consumer, broker configurations, etc.

• The data-bus allowed us to add necessary features that are not available in Kafka (such as topic auditing, monitoring for anomalies in the cluster, creating DLQs, etc.).

• The data-bus allows for centralized failover implementation for all services.

Currently, to start sending events to the message broker, all you need to do is connect a small library into your service’s code. That’s it. You gain the ability to write, read, and scale with just one line of code. The entire implementation is hidden from you, exposing only a few handles such as batch size. Under the hood, the data-bus service spins up the required number of producer and consumer instances in Kubernetes and supplies them with the necessary configuration, all while remaining transparent to your service.

Of course, there is no silver bullet, and this approach has its own limitations.

• Data-bus needs to be supported with your own resources, unlike third-party libraries.
• Data-bus increases the number of interactions between the services and the message broker, which reduces performance compared to raw Kafka.
• Not everything can be easily hidden from the services; duplicating the functionality of KSQL or Kafka Streams in data-bus is not what we want, so sometimes we have to allow services to access directly.

In our case, the benefits outweighed the drawbacks, and the decision to shield the message broker with a separate service proved justified. In a year of operation, we haven’t encountered any serious outages or issues.

P.S. Thanks to my girlfriend, Ekaterina Obalaya, for the cool pictures in this article. If you liked them, here there will be even more illustrations.

Source: habr.com