Understanding Message Brokers. Exploring the mechanics of message exchange through ActiveMQ and Kafka. Chapter 3. Kafka

Continuing the translation of a small book:
«Understanding Message Brokers»,
author: Jakub Korab, publisher: O’Reilly Media, Inc., publication date: June 2017, ISBN: 9781492049296.

Previous translated part: Understanding Message Brokers. Exploring the mechanics of message exchange through ActiveMQ and Kafka. Chapter 1. Introduction

CHAPTER 3

Kafka

Kafka was developed at LinkedIn to overcome some of the limitations of traditional message brokers and to avoid the need to set up multiple message brokers for different point-to-point interactions, as described in this book in the section 'Vertical and Horizontal Scaling' on page 28. Use cases at LinkedIn were primarily based on one-way consumption of very large volumes of data, such as page clicks and access logs, while allowing multiple systems to utilize this data without impacting the performance of producers or other consumers. In fact, the reason Kafka exists is to achieve such a messaging architecture as described by the Universal Data Pipeline.

With this ultimate goal in mind, other requirements naturally arose. Kafka must:

• Be extremely fast
• Provide high throughput when processing messages
• Support 'Publisher-Subscriber' and 'Point-to-Point' models
• Not slow down as more consumers are added. For example, the performance of both queues and topics in ActiveMQ degrades with an increasing number of consumers at the recipient
• Be horizontally scalable; if a single message-persistent broker can only operate at the maximum disk speed, then scaling beyond a single broker instance makes sense to improve throughput
• Separate access to storing and retrieving messages

To achieve all this, Kafka adopts an architecture that redefines the roles and responsibilities of clients and message brokers. The JMS model is heavily broker-oriented, where it is responsible for message dissemination, while clients only need to worry about sending and receiving messages. Kafka, on the other hand, is client-oriented, with clients taking on many functions of the traditional broker, such as fairly distributing relevant messages among consumers, in exchange for an extremely fast and scalable broker. For those who have worked with traditional messaging systems, working with Kafka requires fundamental changes in perspective.
This engineering direction has led to the creation of a messaging infrastructure capable of increasing throughput by several orders of magnitude compared to a conventional broker. As we will see, this approach comes with trade-offs that mean Kafka is not suitable for certain types of loads and established software.

Unified recipient model

To meet the requirements described above, Kafka combines both 'publish-subscribe' and 'point-to-point' messaging types under a single kind of recipient — topic. This confuses people who have worked with messaging systems where the term 'topic' refers to a broadcast mechanism, from which (from the topic) reading is not reliable. Kafka topics should be viewed as a hybrid type of recipient, in accordance with the definition given in the introduction to this book.

For the remainder of this chapter, unless we explicitly state otherwise, the term 'topic' will refer to a Kafka topic.

To fully understand how topics behave and what guarantees they provide, we must first look at how they are implemented in Kafka.
Each topic in Kafka has its own log.
Producers sending messages to Kafka append to this log, while consumers read from the log using pointers that constantly move forward. Periodically, Kafka deletes the oldest parts of the log, regardless of whether the messages in those parts have been read or not. A central part of Kafka's design is that the broker does not care whether messages have been read — that is the client's responsibility.

The terms "log" and "pointer" do not appear in the Kafka documentation. These well-known terms are used here to aid understanding.

This model is completely different from ActiveMQ, where messages from all queues are stored in one log, and the broker marks messages as deleted after they have been read.
Now let's delve a little deeper and take a closer look at the topic log.
The Kafka log consists of several partitions (Figure 3-1). Kafka guarantees strict ordering within each partition. This means that messages written to a partition in a certain order will be read in the same order. Each partition is implemented as a rolling log file that contains a subset (subset) of all messages sent to the topic by its producers. The created topic contains one partition by default. The idea of partitions is a central concept of Kafka for horizontal scaling.

Figure 3-1. Kafka Partitions

When a producer sends a message to a Kafka topic, they decide which partition to send the message to. We will look at this in more detail later.

Reading Messages

The client that wants to read messages manages a named pointer, called consumer group, which points to the offset of the message in the partition. The offset is a position with an increasing number that starts at 0 at the beginning of the partition. This consumer group, referenced in the API through a user-defined identifier group_id, corresponds to one logical consumer or system.

Most messaging systems read data from a topic through multiple instances and streams for parallel message processing. Therefore, there will typically be many consumer instances sharing the same consumer group.

The reading issue can be illustrated as follows:

• A topic has several partitions
• Many consumer groups can simultaneously use a topic
• A consumer group can have several separate instances

This is a non-trivial "many-to-many" problem. To understand how Kafka handles the relationships between consumer groups, consumer instances, and partitions, let’s consider a series of gradually more complex reading scenarios.

Consumers and Consumer Groups

Let’s take a topic with a single partition as a starting point (Figure 3-2).

Figure 3-2. A consumer reads from a partition

When a consumer instance connects to this topic with its own group_id, it is assigned a partition to read from and an offset within that partition. This offset position is configured in the client, pointing to either the most recent position (the newest message) or the earliest position (the oldest message). The consumer polls messages from the topic, resulting in their sequential reading from the log.
The offset position is regularly committed back to Kafka and stored as messages in the internal topic _consumer_offsets. The read messages are not deleted, unlike a typical broker, and the client can rewind the offset to reprocess messages that have already been viewed.

When a second logical consumer connects using a different group_id, it maintains a second pointer that is independent of the first (Figure 3-3). Thus, the Kafka topic acts like a queue, where there is one consumer and, like a typical publish-subscribe topic, multiple consumers subscribed to it, with the added benefit that all messages are retained and can be processed multiple times.

Figure 3-3. Two consumers in different consumer groups read from one partition

Consumers in a consumer group

When one consumer instance reads data from a partition, it fully controls the pointer and processes messages as described in the previous section.
If multiple consumer instances connect with the same group_id to a topic with a single partition, control over the pointer will be given to the instance that connected last, and from that moment on, it will receive all messages (Figure 3-4).

Figure 3-4. Two consumers in the same consumer group read from one partition

This processing mode, where the number of consumer instances exceeds the number of partitions, can be viewed as a form of a monopolistic consumer. This can be useful if you need an "active-passive" (or "hot-warm") clustering of your consumer instances, although the parallel operation of several consumers ("active-active" or "hot-hot") is far more typical than consumers in standby mode.

Such message distribution behavior described above may seem surprising compared to how a standard JMS queue behaves. In this model, messages sent to the queue are evenly distributed among two consumers.

Most often, when we create multiple consumer instances, we do so either for parallel message processing, to increase reading speed, or to enhance the resilience of the reading process. Since only one consumer instance can read data from a partition at the same time, how is this achieved in Kafka?

One way to accomplish this is to use one consumer instance to read all messages and pass them to a thread pool. While this approach increases processing throughput, it complicates the consumer logic and does nothing to improve the resilience of the reading system. If one consumer instance goes down due to a power failure or a similar event, reading stops.

The canonical way to address this issue in Kafka is to use aOlarger number of partitions.

Partitioning

Partitions are the main mechanism for parallelizing reads and scaling the topic beyond the throughput of a single broker instance. To better understand this, let's consider a situation where there is a topic with two partitions and one consumer subscribes to this topic (Figure 3-5).

Figure 3-5. One consumer reads from multiple partitions

In this scenario, the consumer gains control over the pointers corresponding to its group_id in both partitions and starts reading messages from both partitions.
When an additional consumer is added to this topic for the same group_id, Kafka reallocates one of the partitions from the first to the second consumer. Each consumer instance will read from one partition of the topic (Figure 3-6).

To ensure message processing in parallel across 20 threads, you will need at least 20 partitions. If there are fewer partitions, there will be consumers left without anything to work with, as discussed earlier regarding monopolistic consumers.

Figure 3-6. Two consumers in the same consumer group reading from different partitions

This scheme significantly reduces the complexity of how the Kafka broker works compared to the message distribution required to support a JMS queue. There are no concerns about the following aspects:

• Which consumer should receive the next message based on round-robin distribution, current prefetch buffer capacity, or previous messages (as for JMS message groups).
• Which messages have been sent to which consumers and whether they need to be delivered again in case of failure.

All the Kafka broker needs to do is to pass messages to the consumer sequentially when the latter requests them.

However, the requirements for parallelizing reads and resending failed messages do not go away—responsibility for them simply shifts from the broker to the client. This means that they need to be accounted for in your code.

Sending Messages

The responsibility for deciding which partition to send a message to falls on the producer of that message. To understand the mechanism by which this is done, we first need to consider what we are actually sending.

While in JMS we use a message structure with metadata (headers and properties) and a body containing the payload, in Kafka, a message is a pair of 'key-value'. The payload of the message is sent as the value. The key, on the other hand, is mainly used for partitioning and should contain a business-logic-specific key, so that related messages are placed in the same partition.

In Chapter 2, we discussed the online betting scenario where related events need to be processed in order by a single consumer:

1. The user account is set up.
2. Money is credited to the account.
3. A bet is made that withdraws money from the account.

If each event represents a message sent to a topic, then in this case, a natural key would be the account identifier.
When a message is sent using the Kafka Producer API, it is passed to a partitioning function, which, taking into account the message and the current state of the Kafka cluster, returns the partition ID to which the message should be sent. This function is implemented in Java through the Partitioner interface.

This interface looks as follows:

interface Partitioner {
    int partition(String topic,
        Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster);
}

The default implementation of Partitioner to determine the partition uses a key hashing algorithm (general-purpose hashing algorithm over the key) or round-robin if the key is not specified. This default works well in most cases. However, in the future, you may want to write your own.

Writing your own partitioning strategy

Let’s consider an example where you want to send metadata along with the message payload. The payload in our example is an instruction to deposit funds into a gaming account. The instruction is what we want to ensure is not modified during transmission and we want to be certain that only a trusted higher-level system can initiate this instruction. In this case, the sending and receiving systems agree to use a signature for message authenticity verification.
In a typical JMS, we simply define a property called 'message signature' and add it to the message. However, Kafka does not provide us with a mechanism for passing metadata — only a key and a value.

Since the value is the bank transfer payload, the integrity of which we want to maintain, we have no choice but to define a data structure for use in the key. Assuming we need an account identifier for partitioning, since all messages related to an account must be processed in order, we will come up with the following JSON structure:

{
  "signature": "541661622185851c248b41bf0cea7ad0",
  "accountId": "10007865234"
}

Since the signature value will vary depending on the payload, the default hashing strategy of the Partitioner interface will not reliably group related messages. Therefore, we will need to write our own strategy that will analyze this key and partition the accountId value.

Kafka includes checksums to detect message corruption in storage and has a full set of security features. Even so, specific industry requirements sometimes arise, such as the one mentioned above.

The custom partitioning strategy must ensure that all related messages end up in the same partition. While this seems simple, the requirement can be complicated by the importance of ordering related messages and how fixed the number of partitions in the topic is.

The number of partitions in a topic can change over time, as they can be added if traffic exceeds initial expectations. Thus, message keys may be tied to the partition they were initially sent to, implying part of the state that must be distributed among producer instances.

Another factor to consider is the even distribution of messages among partitions. Generally, keys are not distributed evenly across messages, and hash functions do not guarantee fair distribution of messages for a small set of keys.
It is important to note that no matter how you decide to partition messages, the delimiter itself may need to be reused.

Let's consider the requirement for data replication between Kafka clusters in different geographical locations. For this purpose, Kafka comes with a command-line tool called MirrorMaker, which is used to read messages from one cluster and send them to another.

MirrorMaker needs to understand the keys of the replicated topic to maintain the relative order of messages during replication between clusters, as the number of partitions for this topic may not match in both clusters.

Custom partitioning strategies are relatively rare, as the default hashing or round-robin approaches work successfully in most scenarios. However, if you require strict ordering guarantees or need to extract metadata from payloads, partitioning is something you should look into more closely.

The scalability and performance advantages of Kafka are due to offloading some responsibilities of a traditional broker to the client. In this case, a decision is made to distribute potentially related messages across several consumers working in parallel.

JMS brokers also have to deal with such requirements. Interestingly, the mechanism for sending related messages to the same consumer, implemented through JMS Message Groups (a variant of the sticky load balancing strategy), also requires the sender to mark messages as related. In the case of JMS, the broker is responsible for delivering this group of related messages to one consumer out of many and transferring ownership of the group if the consumer fails.

Producer Agreements

Partitioning is not the only thing to consider when sending messages. Let's look at the send() methods from the Producer class in the Java API:

Future  send(ProducerRecord  record);
Future  send(ProducerRecord  record, Callback callback);

It should be noted that both methods return a Future, indicating that the send operation is not performed immediately. As a result, the message (ProducerRecord) is written to the send buffer for each active partition and is sent to the broker in the background thread of the Kafka client library. While this makes operations incredibly fast, it means that an inexperienced application might lose messages if its process is stopped.

As always, there is a way to make the send operation more reliable at the cost of performance. The size of this buffer can be set to 0, forcing the sending application thread to wait until the message is successfully sent to the broker, as follows:

RecordMetadata metadata = producer.send(record).get();

Again, regarding message reading

Reading messages has additional complexities that need to be considered. Unlike the JMS API, which can launch a message listener in response to incoming messages, the interface Consumer only polls (polling). Let’s take a closer look at the method poll (), used for this purpose:

ConsumerRecords  poll(long timeout);

The return value of the method is a container structure consisting of several objects ConsumerRecord from potentially multiple partitions. ConsumerRecord It itself is a holder object for a key-value pair along with corresponding metadata, such as the partition from which it was obtained.

As discussed in Chapter 2, we must constantly remember what happens to messages after their successful or unsuccessful processing, for example, if the client cannot process a message or if it crashes. In JMS, this was handled through the acknowledgement mode. The broker will either delete successfully processed messages or redeliver unprocessed or failed ones (assuming transactions were used).
Kafka operates quite differently. Messages are not deleted in the broker after being read, and the responsibility for what happens in case of failure lies with the reading code itself.

As we have already mentioned, a group of consumers is linked to an offset in the log. The position in the log corresponding to this offset is associated with the next message that will be issued in response to poll ()The moment in time when this offset increases is crucial when reading.

Returning to the reading model discussed earlier, message processing consists of three stages:

1. Extract the message for reading.
2. Process the message.
3. Acknowledge the message.

The Kafka consumer comes with a configuration option enable.auto.commit. This is a commonly used default setting, as is often the case with settings containing the word "auto."

Before Kafka 0.10, a client using this parameter would send the offset of the last read message at the next call poll () after processing. This meant that any messages that had already been fetched could be reprocessed if the client had processed them but was unexpectedly terminated before the call. poll (). Since the broker does not maintain any state regarding how many times a message has been read, the next consumer that fetches this message will not know that something went wrong. This behavior was pseudo-transactional. The offset was only committed if the message was successfully processed, but if the client crashed, the broker sent the same message to another client. This behavior matched the "at least once" message delivery guarantee.at least once«.

In Kafka 0.10, the client code was modified so that commit began to be triggered periodically by the client library in accordance with the setting auto.commit.interval.ms. This behavior lies somewhere between the JMS AUTO_ACKNOWLEDGE and DUPS_OK_ACKNOWLEDGE modes. When using auto-commit, messages could be acknowledged regardless of whether they had actually been processed — this could happen with a slow consumer. If the consumer crashed, messages were fetched by the next consumer starting from the committed position, which could lead to message skipping. In this case, Kafka would not lose messages; the reading code simply did not process them.

This mode has the same prospects as in version 0.9: messages may be processed, but in case of failure, the offset may not be committed, potentially leading to duplicate delivery. The more messages you fetch while performing poll (), the greater this problem.

As discussed in the section "Reading Messages from the Queue" on page 21, there is no concept of exactly once message delivery in the messaging system when considering failure modes.

In Kafka, there are two ways to commit the offset: automatically and manually. In both cases, messages can be processed multiple times if a message was processed but a failure occurred before the commit. You may also not process a message at all if the commit occurred in the background and your code finished before it began processing (possibly in Kafka 0.9 and earlier versions).

You can manage the offset commit process manually in the Kafka consumer API by setting the parameter enable.auto.commit to false and explicitly calling one of the following methods:

void commitSync();
void commitAsync();

If you aim to process a message "at least once," you must manually commit the offset using commitSync(), executing this command immediately after processing messages.

These methods do not allow for acknowledging messages before they have been processed, but they do not do anything to prevent potential duplicate processing, while creating an illusion of transactional behavior. Kafka lacks transactions. The client cannot perform the following:

• Automatically roll back a failed message. Consumers must handle exceptions arising from problematic payloads and backend disconnections themselves, as they cannot rely on message re-delivery by the broker.
• Send messages to multiple topics in a single atomic operation. As we will soon see, control over different topics and partitions may be on different machines in the Kafka cluster, which do not coordinate transactions during sending. As of the time of writing this article, work has been done to make this possible via KIP-98.
• Tie the reading of one message from one topic to sending another message to a different topic. Again, the architecture of Kafka depends on many independent machines working as one bus, and no attempts are made to hide this. For example, there are no API components that would allow linking Consumer and Producer in the transaction. In JMS, this is ensured by the object Session, from which are created MessageProducers and MessageConsumers.

If we cannot rely on transactions, how can we ensure semantics that are closer to what traditional messaging systems provide?

If there is a chance that the consumer's offset may increase before the message has been processed, for example, during a consumer failure, then the consumer has no way of knowing whether its consumer group missed messages when assigned a partition. Thus, one strategy is to rewind the offset to a previous position. The Kafka consumer API provides the following methods for this:

void seek(TopicPartition partition, long offset);
void seekToBeginning(Collection  partitions);

Element.getAnimations() seek () can be used with the method
offsetsForTimes (Map timestampsToSearch) to rewind to a state at a specific point in the past.

Implicitly, using this approach means that it is highly likely that some messages that were previously processed will be read and processed again. To avoid this, we can use idempotent reading, as described in Chapter 4, to track previously viewed messages and exclude duplicates.

As an alternative, your consumer code can be simple if message loss or duplication is acceptable. When we consider use cases for which Kafka is typically employed, such as processing event logs, metrics, click tracking, etc., we understand that losing individual messages is unlikely to significantly affect surrounding applications. In such cases, default values are quite acceptable. On the other hand, if your application needs to process payments, you must carefully handle each individual message. It all comes down to context.

Personal observations indicate that as the intensity of messages increases, the value of each individual message decreases. High-volume messages tend to be valuable when viewed in aggregated form.

High Availability

Kafka's approach to high availability is fundamentally different from that of ActiveMQ. Kafka is designed on horizontally scalable clusters where all broker instances simultaneously receive and distribute messages.

A Kafka cluster consists of multiple broker instances running on different servers. Kafka was developed to work on standard standalone hardware, where each node has its own dedicated storage. The use of network storage systems (SAN) is not recommended, as multiple compute nodes may compete for storage access and create conflicts.IKafka is a continuously running system. Many large Kafka users never shut down their clusters, and the software always ensures updates through sequential restarts. This is achieved by guaranteeing compatibility with the previous version for messages and interactions between brokers.

Kafka is a perpetually active system.

Brokers are connected to the server cluster ZooKeeper, which acts as a configuration data registry and is used to coordinate the roles of each broker. ZooKeeper itself is a distributed system that provides high availability by replicating information through the establishment of a quorum..

In its basic case, a topic is created in the Kafka cluster with the following properties:

• The number of partitions. As discussed earlier, the exact value used here depends on the desired level of parallel reading.
• The replication factor determines how many broker instances in the cluster should contain logs for this partition.

Using ZooKeeper for coordination, Kafka attempts to fairly distribute new partitions among the brokers in the cluster. This is done by a single instance that acts as the Controller.

At runtime, for each topic partition, it assigns a broker to roles of Controller leader (leader, master, primary) and followers. followers (followers, slaves, subordinates). The broker acting as the leader for a given partition is responsible for receiving all messages sent to it by producers and distributing messages to consumers. When messages are sent to a topic partition, they replicate across all broker nodes acting as followers for that partition. Each node containing logs for the partition is called a replica. The broker can act as a leader for some partitions and as a follower for others.

A follower that contains all messages held by the leader is called a synchronized replica (in-sync replica). If the broker acting as the leader for a partition goes down, any broker that is current or synchronized for that partition can take on the role of leader. This is an incredibly resilient design.

A part of the producer's configuration is the parameter acks, which defines how many replicas must acknowledge receipt of a message before the application stream continues sending: 0, 1, or all. If the value is set to all, then upon receiving a message, the leader will send a confirmation back to the producer as soon as it receives acknowledgments of the write from several replicas (including itself) as defined by the topic setting min.insync.replicas (default is 1). If a message cannot be successfully replicated, the producer will raise an exception for the application (NotEnoughReplicas or NotEnoughReplicasAfterAppend).

In a typical configuration, a topic is created with a replication factor of 3 (1 leader, 2 followers for each partition) and the parameter min.insync.replicas is set to 2. In this case, the cluster will allow one of the brokers managing the topic's partition to go offline without affecting client applications.

This brings us back to the familiar trade-off between performance and reliability. Replication incurs additional wait time for acknowledgments from followers. However, since it occurs in parallel, replication to at least three nodes has the same performance as to two (ignoring the increased usage of network bandwidth).

By using this replication scheme, Kafka skillfully avoids the need to ensure the physical recording of each message to disk through the operation sync(). Each message sent by the producer will be recorded in the partition log, but as discussed in Chapter 2, the initial write to file is performed in the operating system's buffer. If this message is replicated to another Kafka instance and resides in its memory, the loss of the leader does not mean that the message itself is lost — its synchronized replica can take over.
The abandonment of the need to perform the operation sync() means that Kafka can accept messages at the rate at which it can write them to memory. Conversely, the longer it can avoid flushing to disk, the better. For this reason, it is not uncommon for Kafka brokers to be allocated 64 GB of memory or more. This memory usage means that a single instance of Kafka can easily operate at speeds thousands of times faster than a traditional message broker.

Kafka can also be configured to apply the operation sync() to batches of messages. Since everything in Kafka is designed to work with batches, it actually works quite well for many use cases and serves as a useful tool for users who demand very strong guarantees. Much of Kafka's raw performance is tied to messages being sent to the broker in batches, and the fact that these messages are read from the broker in sequential blocks using zero-copy operations (operations in which the task of copying data from one memory area to another is not performed). The latter is a significant win in terms of performance and resources and is only possible due to the underlying log data structure that defines the partitioning scheme.

In a Kafka cluster, much higher performance is possible than with a single Kafka broker, as topic partitions can horizontally scale across multiple separate machines.

Summary

In this chapter, we explored how Kafka's architecture redefines the relationship between clients and brokers, ensuring an incredibly resilient messaging pipeline with throughput many times greater than that of a traditional message broker. We discussed the functionality it employs to achieve this goal and briefly reviewed the architecture of the applications that provide this functionality. In the next chapter, we will examine common issues that message-based applications must address and discuss strategies to tackle them. We will conclude the chapter by outlining how to think about messaging technologies in general, so you can assess their suitability for your use cases.

Previous translated part: Understanding Message Brokers. Exploring the mechanics of messaging with ActiveMQ and Kafka. Chapter 1

Translation completed: tele.gg/middle_java

To be continued…

Only registered users can participate in the survey. Please log in, please.

Is Kafka used in your organization?

• Yes

• No

• Used to be, but not anymore

• Planning to use it

38 users voted. 8 users abstained.

Source: habr.com