Chapter 6 rebalance details

rebalance It's an agreement in essence , Set a consumer group All under consumer How to reach an agreement to allocate subscriptions topic Every section of . For example, a certain group There are 20 individual consumer, It subscribes to a with 100 Partitioned topic. Under normal circumstances ,Kafka The average will be for each consumer Distribute 5 Zones . The process of distribution is called rebalance.

6.1 Trigger rebalance Conditions

rebalance There are three trigger conditions for ：

• Changes in group members ( new consumer Join the group 、 existing consumer Take the initiative to leave the group or have consumer collapse )
• The number of subscribed topics has changed —— Of course it is possible , If you use regular expressions to subscribe , Then the new matching regular expression topic It will trigger rebalance
• The number of partitions that subscribe to a topic has changed

6.2 rebalance Allocation mechanism

Kafka The new version consumer Two allocation strategies are provided by default ：range and round-robin. Of course Kafka A pluggable allocation strategy is adopted , You can customize the allocator to implement different allocation strategies . actually , Due to the present range and round-robin Both distributors have some disadvantages ,Kafka The community has proposed a third distributor to achieve a more equitable distribution strategy , It's still under development .

Just a quick example , Suppose that at present some consumer group There are two. consumer：A and B, When the third member joins ,kafka Will trigger rebalance And according to the default allocation policy, it is A、B and C Assign partitions , As shown in the figure below

6.3 perform rebalance and consumer group management

Kafka from Coordinator perform Rebalance and consumer group management ;

First of all 0.8 Version of coordinator, At that time coordinator It's dependence zookeeper To achieve the right consumer group Conduct management .Coordinator monitor zookeeper Of /consumers/<group>/ids Changes of child nodes and /brokers/topics/<topic> Data changes to determine whether it is necessary to rebalance.group Each of the following consumer Decide which partitions to consume , And make your decision first zookeeper Medium /consumers/<group>/owners/<topic>/<partition> Register . Obviously , This scheme depends on zookeeper With the help of the , And each consumer The decision is made alone , No such kind. “ Everyone belongs to a group , Negotiate to do things ” The spirit of .

Based on these potential drawbacks ,0.9 Version of kafka Improved coordinator The design of the , Put forward group coordinator—— Every consumer group Will be assigned one of these coordinator For group management and displacement management . This group coordinator More responsibilities than before , For example, group member management 、 Displacement submission protection mechanism, etc . When the new version consumer group One of the first consumer When it starts , It will kafka server Determine who's in their group coordinator. After that group All the members of the team will be with the coordinator Coordinate communication . Obvious , such coordinator Design is no longer needed zookeeper 了 , The performance can be greatly improved .

6.4 How to determine the Coordinator

that consumer group How to determine your own coordinator Who is it ？ In short, there are two steps

1. determine consumer group Consumption displacement information is written __consumer_offsets Which section of . Specific calculation formula ：

__consumer_offsets_partitionId = Math.abs(groupId.hashCode() % groupMetadataTopicPartitionCount)

Be careful ：groupMetadataTopicPartitionCount By the parameter offsets.topic.num.partitions Appoint , The default is 50 Zones .

2. This partition leader Where broker It's the chosen coordinator

6.5 Rebalance generation

Rebalance generation It said rebalance The next member , Mainly for protection consumer group, Prevent invalidity offset Submit . for example , Last year's consumer Members cannot submit to the new session consumer group in . Otherwise the newspaper ILLEGAL_GENERATION error . Every time group Conduct rebalance after ,generation No. will add 1, Express group Entered a new version , As shown in the figure below ,Generation 1 when group Yes 3 Members , Subsequent members 2 Quit group ,coordinator Trigger rebalance,consumer group Get into Generation 2, Then the members 4 Join in , Trigger again rebalance,group Get into Generation 3.

6.6 agreement （protocol）

rebalance The essence is a set of agreements .group And coordinator Use it together to accomplish group Of rebalance. at present kafka Provides 5 Protocol processing and consumer group coordination Related issues

1. Heartbeat request ：consumer Need to be given regularly coordinator Send a heartbeat to show you're alive
2. LeaveGroup request ： Initiative to tell coordinator I'm leaving consumer group
3. SyncGroup request ：group leader Tell all members of the group about the allocation plan
4. JoinGroup request ： Member requests to join group
5. DescribeGroup request ： Show all information about the group , Include member information 、 Name of agreement 、 Distribution plan 、 Subscription information, etc . Usually the request is for the administrator to use

Coordinator stay rebalance I mainly used the front 4 Kind of request .

6.7 Liveness

consumer How to talk to coordinator Prove you're alive ？ By timing to coordinator send out Heartbeat request . If the set timeout is exceeded , that coordinator Think of this as consumer It's gone . once coordinator Think of someone as consumer Hang up , Then it will start a new round rebalance, And in the current other consumer The heart of response Add “REBALANCE_IN_PROGRESS”, Tell others consumer： Sorry, guys , You re apply to join the Group ！

6.8 Rebalance The process

rebalance The premise is coordinator It's confirmed . Overall speaking ,rebalance It is divided into 2 Step ：Join and Sync

1. Join： As the name suggests, join the group . All the members turned to coordinator send out JoinGroup request , Request to join the group . Once all members have sent JoinGroup request ,coordinator Will choose one of them consumer As a leader Role , And send group member information and subscription information to leader—— Be careful leader and coordinator It's not a concept .leader Responsible for the formulation of consumption distribution plan .
2. Sync：leader Start to allocate spending programs , Which one consumer What is responsible for consumption topic Which of partition. Once the distribution is complete ,leader Will encapsulate this solution into SyncGroup To send to in a request coordinator, Not leader I can also send SyncGroup request , It's just that the content is empty .coordinator After receiving the allocation scheme, the scheme will be crammed into SyncGroup Of response To each consumer. In this way, all members of the group will know which partitions they should consume .
   

Be careful ： stay coordinator Before all member requests are collected , It will put the received request into a file called purgatory( Purgatory ) The place of .

Then there is the process of distributing the distribution scheme , namely SyncGroup request ：

Be careful ：consumer group The partition allocation scheme is implemented on the client side ！Kafka The main reason for delegating this right to the client is that it allows for better flexibility . For example, under this mechanism, we can achieve something similar to Hadoop That kind of rack perception (rack-aware) Distribution plan , That is to say consumer Select partition data under the same rack , Reduce the overhead of network transmission . You can override consumer Parameters of partition.assignment.strategy To implement your own distribution strategy .

6.9 Consumer group State machine

consumer group A state machine is also made to indicate the flow of group state .coordinator According to this state, the opportunity is right consumer group Do different things , As shown in the figure below

Briefly explain the states in the figure below

• Dead： The final status of no member in the group , The metadata of the group has also been coordinator Removed . This state is a response to various requests response： UNKNOWN_MEMBER_ID
• Empty： There are no members in the group , But the displacement information hasn't expired yet . This state can only respond to JoinGroup request
• PreparingRebalance： The group is ready to open a new rebalance, Waiting for members to join
• AwaitingSync： Is waiting for leader consumer Pass the allocation scheme to each member
• Stable：rebalance complete ！ You can start spending ~

6.10 Rebalance Scene analysis

6.10.1 New members join the group

6.10.2 Group members crash

There are two different scenarios for group members to crash and group members to leave actively . Because in the event of a crash, members don't tell coordinator The matter ,coordinator There may be a need for a complete session.timeout Cycle can detect this kind of collapse , This is bound to cause consumer The lag of . It can be said that to leave the group is to initiate rebalance; And the collapse is passive rebalance.

6.10.3 Group members voluntarily leave the group

6.10.4 Submit displacement