[Kafka] - 10. Consumer_Rebalance

Partition-Assignment

---

하나의 Partition은 지정된 Consumer_group 내에 하나의 Consumer만 사용 가능

동일한 Key를 가진 Message는 동일한 Consumer가 사용한다.

Consumer의 설정 Parameter 중 Partition.assignment.strategy로 할당 방식을 변경이 가능하다.

Consumer Group은 Group Coordinator 프로세스에 의해 관리 된다.

 

Partition.Assignment.Strategy

• org.apache.kafka.clients.consumer.RangeAssginor : Topic별로 작동하는 Default Assignor
• org.apache.kafka.clients.consumer.RoundRobinAssignor : Round Robin 방식으로 Consumer에게 Partition을 할당한다.
• org.apache.kafka.clients.consumer.StickyAssginor : 최대한 많은 기존 Partition 할당을 유지하면서 최대 균형을 이루는 할당을 보장한다.
• org.apache.kafka.clients.consumer.CooperativeStickyAssignor : 동일한 StickyAssignor 논리를 따르지만 협력적인 Rebalance를 허용한다.
• org.apache.kafka.clients.consumer.ConsumerPartitionAssginor : 인터페이스를 구현하면 사용자 지정 할당 전략을 사용할 수 있다.

 

Topic A에 대해 3개의 Broker와 6개의 Partition과 7개의 Consumer가 있으며, Topic의 Partition 갯수는 Default=50이라고 가정하여 보자

 

1. Consumer 등록 및 Group Coordinator 선택

• 각 Consumer는 Group.id를 통해 Kafka Cluster로 자신을 등록한다.
• Kafka는 Consumer_Group을 만들고 Consumer에 모든 Offset을 __Consumer_offsets의 하나의 Partition에 저장한다.
• 해당 Partition의 Leader Broker는 Consumer Group의 Group Coordinator로 선택(알고리즘에 의해)

Hash(Group.id)%offsets.topic.num.partitions

알고리즘을 통하여 해당 값을 가지고 있는 Leader Partition의 Broker가 Group Coordinator

 

 

2. Join Group 요청에 따라 Consumer 나열

• Consumer들은 Boker에 붙기 위해 Join Group을 날린다.
• 이 때, 대기 시간이 group.initial.rebalance.delay.ms(default 3초) 를 대기
• Consumer0~6번은 Group Coordinator에게 Join Group을 날린다.

 

3. Group Leader 결정 및 Partition 할당

• Join Group 요청을 하는 최초의 Consumer는 Group Leader가 된다.
• Group Leader는 Group Coordinator로 부터 Consumer목록을 받는다.
• Group Leader는 구성된 Partition.assignment.startegy를 사용하여 각 Consumer에게 Partition을 할당한다.
• 아래 그림에서는 Partition보다 Consumer가 많기 때문에 Consumer6은 할당 받지 못하였다.
• Consumer보다 Partition의 갯수가 더 많다면, Consumer에 여러개의 Parition이 할당된다.

 

 

4. Consumer:partition mapping info를 Group Coordinator에게 전달

• Group Leader는 Mapping Info를 Group coordinator에게 보낸다.
• Group Coordinator는 Mapping 정보를 Cache하여 Zookeeper에 유지

 

5. 각 Consumer에게 할당된 Parition 정보를 보낸다.

• Group Coordinator는 각 Consumer에게 할당된 Partition 정보를 보낸다.
• 각 Consumer들은 할당된 Partition으로 부터 Consume 시작
• Consumer6은 할당 받은 Partition이 없음으로 노는 상태.

 

 

지금까지 Consumer들이 Consumer Group을 구성하고 Broker와 함께 Partition을 할당 받고 Mapping하여 처리하는 모습을 보았다. 여기서 의문이 드는 점은

왜 Group Coordinator가 직접 Consumer들을 Mapping하지 않고 Consumer 중 하나인 Group leader가 이를 Mapping하는가?

Kafka는 가능한 계산 처리를 Client(Producer,Consumer)에게 수행하도록 하여 Broker에게 부담을 줄여주도록 Architecture를 구성중에 있다.

만약 Consumer Group과 Consumer가 엄청 나게 많다면.. Broker가 1개인 경우에 Broker는 계산 처리하는데 부담을 느낀다.

 

Consumer Rebalancing Trigger

---

불필요한 Rebalancing을 피해야 한다. ( Consumer들은 Rebalancing 시에 Message를 Consume하지 못하여, 계속 발생하면 영향을 받는다. )

Rebalancing Trigger Case

• Consumer가 Consumer Group에서 탈퇴
• New Consumer가 Consumer Group에 합류
• Consumer가 Topic을 변경
• Consumer Group은 Topic Metadata 변경인지( Partition 갯수 변화.. ) 

 

Consumer HealthChek

---

Consumer의 장애를 인지하기 위한 용도

• Consumer는 Poll()과 별도로 백그라운드 Thread에서 Heartbeat를 보낸다.

# default=3초
Heartbeat.interval.ms

• 일정 시간 동안 Heartbeat를 수신하지 못하면, Consumer는 Consumer Group에서 삭제된다.

# default=10초
Session.timeout.ms

• Poll()은 Heartbeat와 상관 없이 주기적으로 호출되어야 한다.

# default=5분
Max.poll.interval.ms

 

과도한 Rebalancing 방지(성능 최적화)

---

1. Consumer Group Member 고정

Group의 각 Consumer에게 고유한 Group.instance.id를 할당

Consumer는 Leave Group Request를 사용하지 않는다.(사용하게 되면 Group에서 떠나게 설정한다는 의미.. )

Rejoin은 알려진 Grooup.instance.id에 대해서는 Rebalancing Trigger를 발생시키지 않는다.

 

2. Session.timeout.ms 튜닝

heartbeat.interval.ms를 session.timeout.ms의 1/3로 설정

Group.min.session.timeout.mx(default=6초)와 Group.max.session.timeout.mx(default=5분)의 사이값으로 설정

장점 : Consumer가 Rejoin할 시간적 여유를 준다

단점 : Consumer의 장애를 인지하는 시간이 delay된다.

 

3. Max.poll.interval.ms 튜닝

Consumer에게 Poll()한 Data를 처리 할 수 있는 충분한 시간 제공( 코드에 맞추어서 테스트해봐야한다. )

너무 크게 하면 안된다.