Where does Kafka have the concept of partition allocation? Describe the general process and principles

When a user sends a message using the Kafka client, the send method is called and the message is automatically sent to the Broker.

This process needs to go through an interceptor, serializer, partitioner, and so on before it can actually be sent to the broker. A message needs to identify the partition to which it needs to be sent before it can be sent to the broker. If a partition field is specified in ProducerRecord, there is no need for a partition because partition represents the partition number to be sent to. If no partition field is specified in the ProducerRecord message, you need to rely on the partition divider to calculate the value of the partition based on the key field. The purpose of a divider is to partition messages.

The default partition provided in Kafka is DefaultPartitioner, which implements the Partitioner interface (the user can implement this interface to customize the partition). The partition method is used to implement the specific partition allocation logic:

public int partition(String topic, Object key, byte[] keyBytes,
                     Object value, byte[] valueBytes, Cluster cluster);
Copy the code

By default, if the key of the message is not null, the default partition partition hashes the key (using the MurmurHash2 algorithm with high performance and low collision rate), eventually calculates the partition number based on the hash value, and messages with the same key are written to the same partition. If the key is null, the message will be polled to the various availability zones of the topic.

Note: If the key is not null, the computed partition number will be any one of the partitions. If the key is null and there are availability zones, the partition number calculated is either of the availability zones. Note the difference between the two.

The principle of the RangeAssignor strategy is to divide the total number of consumers by the total number of partitions to obtain a span, and then allocate the partitions evenly across the span to ensure that the partitions are distributed as evenly as possible to all consumers. For each topic, the RangeAssignor strategy sorts all the consumers in the consumer group that subscribe to that topic in lexicographical order by name, and then allocates a fixed partition range for each consumer. If the distribution is not equal, the highest consumer in the lexicographical order is assigned an additional partition.

Assuming n= number of partitions/number of consumers, and m= number of partitions % number of consumers, then the first m consumers are allocated N +1 partitions each, and the following (number of consumers -m) consumers are allocated N partitions each.

To illustrate the RangeAssignor strategy in a more general way, let’s give a few more examples. Suppose there are two consumers C0 and C1 in the consumer group, both subscribed to topics t0 and T1, and each topic has four partitions. Then all the partitions subscribed to can be identified as t0P0, T0P1, T0P2, T0P3, T1P0, T1P1, T1P2, and T1P3. The final distribution result is:

Consumer C0: T0P0, T0P1, T1P0, T1P1 Consumer C1: T0P2, T0P3, T1P2, T1P3Copy the code

This distribution is very uniform, but can this distribution strategy maintain this good property? Let’s look at another case. Assuming that both topics in the above example have only three partitions, all subscribed partitions can be identified as t0P0, T0P1, T0P2, T1P0, T1P1, t1P2. The final distribution result is:

Consumer C0: T0P0, T0P1, T1P0, T1P1 Consumer C1: T0P2, T1P2Copy the code

It is obvious that the distribution is not even. If the similar situation is enlarged, some consumers may be overloaded. Let’s look at another round Bin Assignor strategy.

Take a look at StickyAssignor. The word sticky can be translated as sticky. Kafka has introduced sticky assignor since version 0.11.x.

1. Partition distribution should be as uniform as possible;
2. Partitions are allocated as much as possible as they were last allocated. When the two conflict, the first goal takes precedence over the second. Given these two goals, StickyAssignor is much more complex to implement than the RangeAssignor or RoundRobinAssignor. Let’s take a look at the StickyAssignor strategy in action.

Suppose there are three consumers in the consumer group: C0, C1, and C2, each of which subscribes to four topics: T0, T1, T2, and T3, and each topic has two partitions, that is, the entire consumer group subscribes to eight partitions: T0P0, T0P1, T1P0, T1P1, T2P0, T2P1, T3P0, and T3P1. The final distribution result is as follows:

Consumer C0: T0P0, T1P1, T3P0 Consumer C1: T0P1, T2P0, T3P1 Consumer C2: T1P0, T2P1Copy the code

This may at first appear to be the same result as a round Bin Assignor, but is it really the case? Assuming that consumer C1 is separated from the consumer group, the consumer group will perform a rebalancing operation, and the consumption partition will be reallocated. If RoundRobinAssignor is used, the assignment results are as follows:

Consumer C0: T0P0, T1P0, T2P0, T3P0 Consumer C2: T0P1, T1P1, T2P1, T3P1Copy the code

As shown in the allocation results, the RoundRobinAssignor policy repolls the allocation for consumers C0 and C2. If StickyAssignor is used, the result is:

Consumer C0: T0P0, T1P1, T3P0, T2P0 Consumer C2: T1P0, T2P1, T0P1, T3P1Copy the code

It can be seen that the allocation result retains all the allocation results of the previous allocation for consumers C0 and C2, and distributes the “burden” of the original consumer C1 to the remaining two consumers C0 and C2. Finally, the allocation of C0 and C2 remains balanced.

If partition reallocation occurs, it is possible that the previous consumer and the newly assigned consumer are not the same for the same partition, and that the halfway processing of the previous consumer has to be repeated in the newly assigned consumer, which is obviously a waste of system resources. StickyAssignor, like the sticky in its name, makes the allocation policy sticky and tries to make two assignments as same as possible, thus reducing the loss of system resources and other exceptions.

So far, we have analyzed the same subscription information of consumers. Let’s take a look at the processing of different subscription information.

For example, there are three consumers in the same consumer group: C0, C1, and C2, and there are three topics in the cluster: T0, T1, and T2, which have 1, 2, and 3 partitions respectively. That is to say, there are six partitions in the cluster: T0P0, T1P0, T1P1, T2P0, T2P1, and T2P2. Consumer C0 subscribes to topics T0, consumer C1 to topics t0 and T1, and consumer C2 to topics t0, T1, and T2.

If RoundRobinAssignor is used, the result will look like this:

[Allocation Result Set 1] Consumer C0: T0P0 Consumer C1: T1P0 Consumer C2: T1P1, T2P0, T2P1, t2P2Copy the code

If StickyAssignor is used, the final allocation is as follows:

[Allocation Result Set 2] Consumer C0: T0P0 Consumer C1: T1P0, t1P1 Consumer C2: T2P0, T2P1, t2P2Copy the code

As you can see, this is an optimal solution (consumer C0 does not subscribe to topics T1 and T2, so it cannot be assigned any partitions in topics T1 and T2, and the same can be inferred for consumer C1). If consumer C0 breaks away from the consumer group, the allocation result of RoundRobinAssignor strategy is:

Consumer C1: T0P0, T1P1 Consumer C2: T1P0, T2P0, T2P1, t2P2Copy the code

As you can see, the RoundRobinAssignor strategy preserves the original allocation of three partitions for consumers C1 and C2: T2P0, T2P1, and T2P2 (for result set 1). If StickyAssignor is used, the result is:

Consumer C1: T1P0, T1P1, T0P0 Consumer C2: T2P0, T2P1, t2P2Copy the code

As you can see, StickyAssignor preserves the existing allocation of five partitions for consumers C1 and C2: T1P0, T1P1, T2P0, T2P1, and T2P2.

As a result, StickyAssignor is much better than the other two, and the code implementation of this strategy is extremely complex.

If there are two allocation policies configured in the consumer client, which one will prevail? If there are multiple consumers with different allocation policies configured for each other, which one shall prevail? The partition allocation between multiple consumers needs to be coordinated, so what is the process of this coordination?

A GroupCoordinator in kafka coordinates the partition allocation of each consumer group. For each consumer group, there is a GroupCoordinator on the kafka server. The process for coordinating zone allocation is as follows: 1. Each consumer submits its own allocation policy to a GroupCoordinator. As shown in the figure below, the allocation policy and subscription information for each consumer proposal is included in the JoinGroupRequest request.

2. A GroupCoordinator collects proposals from each consumer and performs the following two steps: 1. Second, the partition allocation strategy of election consumption group.

It is easy to understand why the leader of the consumer group should be elected, because the final implementation of the partition allocation strategy needs to be executed by a member, and the leader consumer just plays this role. In Kafka, enforcement of the specific partition allocation policy is delegated to the consumer client, which provides greater flexibility. For example, if the allocation policy needs to be changed, the consumer client can wake up by modifying it, rather than modifying and restarting the Kafka server.

How to elect the leader of the consumption group? This can be analyzed in two cases: if there is no leader in the consumption group, the first consumer who joins the consumption group is the leader of the consumption group; If the leader consumer quits the consumer group at some point due to some reasons, a new leader will be elected, and the process of re-electing the leader is more “arbitrary”, the relevant code is as follows:

//scala code.
private val members = new mutable.HashMap[String, MemberMetadata]
var leaderId = members.keys.head
Copy the code

To explain these two lines of code: In a GroupCoordinator, consumer information is stored as a HashMap, where the key is the name of the consumer and the value is consumer-related metadata information. The leaderId represents the name of the Leader consumer, and its value is the key of the first key-value pair in the HashMap, which is basically the same as a random selection. Generally speaking, the leader election process of the consumer group is very random.

How to elect the allocation strategy of the consumer group? Take a vote. Each consumer can set its own partition allocation strategy. For the consumer group, it is necessary to elect a mutually “convincing” strategy from each distribution strategy submitted by each consumer to carry out the overall partition allocation. Elections for this partition allocation are not determined by the Leader consumers, but by the votes of individual consumers within the consumer group. In this case, the “decision is made according to the votes of each consumer in the group” does not mean that the GroupCoordinator further interacts with each consumer to implement the decision. Instead, the GroupCoordinator implements the decision according to the allocation strategy submitted by each consumer. The final distribution strategy can basically be regarded as the one most supported by each consumer. The specific election process is as follows:

Collect all allocation strategies supported by each consumer and form candidate candidates.

Each consumer identifies the first strategy he/she supports from the candidates and votes for this strategy.

Calculate the votes of each strategy in the candidate set. The strategy with the most votes is the allocation strategy of the current consumer group.

If a consumer does not support the elected allocation strategy, an error is reported.

3. The GroupCoordinator sends a receipt to each consumer and assigns a zone to the leader consumer.

As shown in the figure above, the JoinGroupResponse receipt contains information about the allocation policy that is voted by the GroupCoordinator. Moreover, only the leader consumer’s receipt contains the subscription information of each consumer, because only the Leader consumer needs to perform the specific allocation based on the subscription information, the rest of the consumption does not need.

4. After sorting out a specific partition allocation scheme, the leader consumer requests SyncGroupRequest and submits it to the GroupCoordinator. The GroupCoordinator then selects the distribution result for each consumer and informs them via SyncGroupResponse receipt.