Kafka 심화개념 - 1

2-1. Producer Acks, Batch, Page Cahche, Flush

 

프로듀서는 Kafka가 메시지를 잘 받았는지 어떻게 알 수 있는가?

acks: 요청이 성공하는 때를 정의하는 데 사용되는 producer에 설정하는 parameter

 

acks=0

> ack가 필요하지 않음. 메시지 손실이 있더라도 빠르게 메시지를 보내야하는 경우 사용됨

> 응답 없음

 

acks=1 (default값)

> leader가 메시지를 받으면 producer에게 ack를 보냄

> ack를 보내고 follower가 복제 전에 leader에 장애 발생 시, 메시지 손실됨

> At most once 최대 한 번 전송 보장

 

acks=-1 / acks=all

> leader가 모든 replica까지 Commit되면 ack 보냄

> 메시지 손실 방지

> 대기시간이 길고 특정 실패 사례에서 반복되는 데이터 발생 가능성이 있음

> At least once 최소 한 번 전송 보장

 

Producer Retry: 재전송을 위한 parameter

 

Retries 조정 대신 delivery.timeout.ms 조정으로 재시도 동작제어

acks=0에서 retry는 무의미함

 

Producer Batch 처리: 메시지를 모아 한 번에 전송

> RPC(Remote Producer Call) 수를 줄여서 Broker 처리 작업 줄어들어 더 많은 처리량 제공

 

linger.ms (default: 0, 즉시 보냄)

> 메시지가 함께 배치 처리될 때까지 대기 시간

 

batch.size (default:16KB)

> 보내기 전 배치의 최대 크기

 

일반적인 설정: linger.ms=100, batch.size=1000000

 

Producer Delivery Timeout: send() 후 성공 또는 실패를 보고하는 시간의 상한

 

Message Send 순서 보장

> 진행 중 (in-flight)인 여러 요청(request)을 재시도하면 순서가 변경될 수 있음

메시지 순서 보장하려면 Producer에서 enable.idempotence=true 설정

 

enable.idempotence=true 설정 시,

하나의 배치 실패하면 같은 partition으로 들어오는 후속 배치들도 OutOfOrderSequenceException 오류 발생시킴

 

Page Cache와 Flush

Partition은 log Segment file로 구성 (1GB마다 새로운 Segment 생성)

성능 위해 Log Segment는 OS Page Cache에 기록

로그 파일 저장된 메시지 데이터 형태 =  Broker가 Producer로부터 수신한 것 = Consumer에게 보내는 것 >> Zero-Copy 가능

Page Cache가 디스크로 Flush 되는 경우

> Broker가 완전히 종료

> OS Background “Flush Thread” 실행

 

Zero-copy: 데이터가 User Space 복사 없이 CPU 개입 없이 Page Cache와 Network Buffer 사이 전송. Broker Heap 메모리 절약 및 엄청난 처리량 제공

 

Flush 전에 Broker 장애 시,

Replication되어 있으면 Broker가 온라인 상태 될 때 Leader Replica에서 데이터 복구

복제 없으면 데이터 소실됨

 

kafka는 운영체제의 Background flush 기능을 허용하고 이 설정을 기본값으로 유지하는 걸 권장

 

*.log 파일에 디스크로 flush된 데이터와 아직 flush 안 된 page cache 데이터 모두 표시됨

linux의 vmtouch로 flush된 항목과 되지 않은 항목 표시 가능

 

2-2.Replica Failure

 

Follower 실패 케이스

> Leader에 의해 ISR 리스트에서 삭제됨

> Leader는 새로운 ISR 사용하여 Commit

 

n개의 replica가 있는 경우 n-1개의 장애 발생 가능

 

ISR은 Leader가 관리함

Zookeeper에 ISR 업데이트, Controller가 Zookeeper로부터 수신

 

Follower가 느리면 Leader는 ISR에서 Follower를 제고하고 Zookeeper에 ISR유지

(replica.lag.time.max.ms 이내 Follower가 fetch 안 할 경우 ISR에서 제거)

Controller는 파티션 메타데이터에 대한 변경사항에 대해 Zookeeper로부터 수신

 

Leader 실패하면,

Zookeeper가 장애 감지(session timeout)하면 Controller가 새로운 leader 선출 및 새로운 ISR 리스트를 Zookeeper에 기록

Controller가 모든 broker에게 새로운 ISR Push

Client(Consumer,Producer)는 메타데이터 요청하여 새로운 Leader 확인

 

파티션 생성 시 Broker들 사이에서 파티션들 분산 배치됨

(Leader와 Follower가 특정 Broker에 몰리지 않도록)

 

파티션에 Leader가 없으면 Leader 선출될 때까지 해당 파티션 사용 못함

Producer의 send()는 retries 파라미터가 설정되면 재시도

retries=0이면 Network 오류

 

2-3. Replica Recovery

 

acks=all이 굉장히 중요! Leader 장애가 나더라도 Producer가 ack 받을 때까지 메시지 송신하므로, Follower에 데이터 적재 가능하며 Leader 선출 가능. 단, 데이터 중복이 있을수 있음

acks=1이면, Leader에서 수신되면 바로 ack주기 때문에, 장애 시 follower가 수신된 값 모두를 못 받아옴. Producer는 ack 받았기에 재송신 하지 않기 때문.

 

Availability(가용성)과 Durability(내구성)을 선택할 수 있는 파라미터

 

2-4. Consumer Rebalance

 

동일 토픽에서 Consume하는 여러 Consumer Group이 있을 수 있음 (여러개의 Consumer Group)

 

파티션을 Consumer에게 Assign(할당) 과정

하나의 파티션은 지정된 Consumer Group내의 하나의 Consumer만 사용

동일한 Key를 가진 메시지는 동일한 Consumer가 사용(파티션 수를 변경하지 않는 한)

Consumer의 설정 파라미터 중 Partition.assignment.strategy로 할당 방식 조정

Consumer Group은 Group Coordinator라는 프로세스로 작동

 

Group Coordinator(하나의 Broker)와 Group leader(하나의 Consumer) 상호작용

1) Consumer 등록 및 Group Coordinator 선택

2) JoinGroup 요청 순서에 따라 Consumer 나열

3) Group Leader 결정 및 파티션 할당

4) Consumer->Partition 맵핑정보를 Group Coordinator에게 전송

5) 각 Consumer에게 할당된 파티션 정보를 보냄

• Consumer보다 파티션이 적은 경우, 파티션이 할당되지 않은 Consumer는 Idle(쉼)

 

왜 Group Coordinator(Broker)가 직접 파티션을 할당하지 않는지

> kafka의 원칙: 가능한 한 많은 계산을 클라이언트에 수행하도록 하여 Broker의 부담 줄이기

> Consumer 양이 많아지면 Broker에 부담이 커짐

 

Consumer Rebalancing Trigger

불필요한 리밸런싱은 피하자

> Consumer가 Consumer Group 탈퇴

> 신규 Consumer가 Consumer Group 합류

> Consumer가 토픽 구독 변경

> Consumer Group은 토픽 메타데이터 변경사항을 인지함(ex, 파티션증가)

 

리밸런싱 프로세스

> Group Coordinator는 heartbeats의 플래그 사용하여 Consumer에게 리밸런싱 신호 송신

> Consumer가 일시 중지하고 오프셋을 커밋

> Consumer는 Consumer Group의 새로운 Generation에 합류

> 파티션 재할당

> Consumer는 새 파티션에서 다시 Consume 시작

 

** Consumer Rebalancing 시에 Consumer들은 메시지 Consume 불가하므로, 꼭 필요할 때에만 리밸런싱 하기!

 

Consumer Heartbeats

: Consumer 장애를 인지하기 위한 신호

 

Consumer는 poll()과 별도로 백그라운드 Thread에서 Heartbeats 보냄 (heartbeat.interval.ms: default 3s)

Session.timeout.ms(default 10s) 시간동안 heartbeats 미수신 시, Consumer는 Consumer Group에서 삭제

poll()은 Heaertbeats와 상관없이 주기적으로 호출되어야 함(max.poll.interval.ms: default 5min)

 

과도한 리밸런싱 피하는 법

> Consumer Group 멤버 고정(각 Consumer에게 group.instance.id 할당)

> session.timeout.ms 튜닝

> max.poll.insterval.ms 튜닝

 

2-5. Partition Assignment Strategy

 

Consumer 파라미터 중 partition.assignment.strategy로 할당 방식 조정 가능

 

Range Assignor방식: 기본 Assignor

 

라운드 로빈 방식: Range 방식보다 효율적으로 분배하여 할당

> Consumer간 Subscribe 해오는 토픽이 다르면, 할당 불균형이 발생할 수 있음

 

Sticky Assignor 방식

1)가능한한 균형적으로 할당 보장

> Consumer에게 할당된 토픽 파티션 수는 최대 1개

2)재할당 발생 시, 기존 할당을 최대한 많이 보존하여 유지

> 토픽 파티션이 하나의 Consumer에서 다른 Consumer로 이동할 때(리밸런싱) 오버헤드 줄임

 

 

2-6. Cooperative Sticky Asignor

 

시간 흐름에 따른 Consumer Rebalance 과정

 

Eager Rebalance 프로토콜

: 최대한 단순하게 유지하기 위해 만들어짐

 

Incremental Cooperative Rebalancing Protocol

: Eager Rebalancing 프로토콜 보다 발전한 방식

> Revoke할 파티션만 Revoke 처리하고 새로 할당할 것만 해주기

> 이상적인 Consumer Rebalancing 프로토콜

> Rebalancing을 두 번 수행

> 1st rebalance에서 Consumer는 자신의 파티션 중 어느 것이 다른 곳으로 재할당되어야 하는지 알게 됨

> Apache Kafka 2.5에서 추가됨

> 빈번하게 리밸런싱되거나 스케일 인/아웃으로 인한 다운타임 우려 시 2.5 이상 기반으로 권장

(2.4까지는 Basic Rebalancing Protocol 사용하게 됨)