삐끄덕거리며 조립되어 가는

| Kafka란?

• 카프카(Kafka)는 고성능 분산 이벤트 스트리밍 플랫폼이다. 흔히 메시지(이벤트) 큐라고 불리우며 Pub-Sub 모델 형태로 동작하며 분산환경에 특화되어 있다.
• producer 라고 불리우는 source application으로 부터 전송 받은 메시지를 broker에 저장해두며 이를 consumer라는 target application에서 사용할 수 있도록 해준다.
• kafka는 pub/sub 모델로써 consumer가 broker에 있는 메시지를 polling하는 형태로 broker의 부하를 줄여주어 효율적으로 대용량 처리를 병렬로 가능하게끔 한다.
• 메시지를 파일 시스템에 저장하기 때문에 데이터의 영속성을 보장한다.
• 기존의 메시징 시스템은 메시지 큐에 적재된 메시지 양이 많을수록 성능이 크게 감소하였지만, Kafka는 메시지를 파일 시스템에 저장하기 때문에 성능이 크게 감소하지 않는 이점이 있다.

| 주요 내용 정리

• Producer
  • 데이터를 생성하여 kafka cluster에 데이터를 전송하는 역할을 한다.
  • kafka는 topic이라는 주제에 따라 데이터를 전송하고 broker는 topic을 기준으로 구분하여 데이터를 저장한다.
  • 설정에 따라 재처리 또는 실패에 대해 관리가 가능하다. ( DLQ = Dead Letter Queue 로 관리 할 수도 있다.)
  • kafka는 key를 고유 로직으로 hashing 처리하여 저장하기에 동일한 key는 동일한 파티션에 항상 담기게 된다.
  • But! 파티션이 늘어난다면(변동이 생긴다면) 이 rule이 깨지기 때문에 rule을 유지해야 하는 서비스라면 최초에 설정시 파티션의 갯수도 적정하게 설정할 필요가 있다.
• Broker
  • 데이터 저장 : producer로 부터 전송받은 데이터를 영속적으로 저장한다. 저장시에는 topic별로 , 그리고 토픽내 파티션에 분산하여 저장한다.
  • producer로 받은 데이터를 consumer가 가져갈 수 있게 역할을 수행. consumer는 토픽별로 데이터를 읽어가서 처리가 가능하다.
  • kafka cluster 관리 : broker는 cluster의 상태와 구성을 관리한다. 파티션 중 리더 파티션을 선출하고 레플리카의 데이터 복제를 관리.
• Topic
  • 메시지를 저장하는 논리적인 그룹의 단위
  • RDB로 생각하면 각각의 테이블 처럼 데이터를 저장함.
  • 데이터의 형태는 다양한 형태로 저장 가능 ( ex. string, set, json 등등)
  • 하나의 토픽은 보통 1개 이상의 파티션으로 구성되어 분산 저장(처리)이 되는 형태.
  • 토픽에 저장된 데이터는 consumer가 소비해도 삭제되지 않는다. ( 고가용성이 가능한 이유 중 하나다.)
  • 메시지 삭제 정책은 존재하여 시간이나 데이터 총 사이즈로 설정할 수 있다.
• Partition
  • 토픽은 하나의 브로커에 하나의 논리적인 곳에 저장되지 않고 데이터를 파티션으로 나누어서 저장한다.
  • 라운드-로빈 방식으로 데이터를 각 파티션에 저장한다.
  • 파티션이 나뉘어져 있어서 병렬로 데이터의 비동기 처리가 가능하다.
  • 파티션이 2개 이상이면 데이터 처리의 순서 보장은 불가능하다.
• Consumer
  • kafka broker에 topic에 저장된 데이터를 subscribe하여 그에 맞게 처리하는 역할을 한다.
  • consumer는 보통 consumer group을 구성하여 그룹내에서 파티션을 각각 맵핑 및 접근하여 처리한다.
• Consumer group
  • consumer group은 특정 토픽을 동일하게 처리할 때 병렬로 분산 처리하기 위해서 그룹핑된 컨슈머 집합.
    consumer group에 속한 각각의 컨슈머는 하나의 토픽에 여러개의 파티션이 있을 때 같은 파티션에는 하나의 컨슈머만 접근이 가능하도록 제한이 되어 있다. 하지만 하나의 컨슈머는 하나의 토픽에 속한 여러개의 파티션에는 중복적으로 접근하여 처리가 가능하다.
  • 컨슈머 그룹내에는 group leader가 존재하여 consumer가 이슈가 있거나 할 때 리밸런싱을 하는 등의 관리를 맡는다. reader consumer는 토픽에 최초로 ack를 보내어 응답을 받은 consumer가 자동으로 할당된다.
• *** 컨슈머 그룹과 파티션 맵핑 예시 ****

• 1] join group 호출 (내가 너의 토픽을 consume 할거야!)
• 2] group leader 선출 (선착순!!)
• 3] 파티션과 컨슈머 그룹내 consumer와의 맵핑
• 4] 해당 맵핑 정보 전송 ( Group coordinator는 해당 정보를 zookeeper에게 전달)
• 5] 컨슈머에게 해당 내용 전파 ( 전파된 파티션으로 부터 데이터 consume 진행)

** 파티션과 컨슈머 갯수별 맵핑은??

• 1) 토픽 파티션 1 : 컨슈머 1 => 이렇게 구성할 경우 순서의 보장은 될 수 있으나 컨슈머가 문제가 생길경우 메시지를 처리할 수 없다.(장애발생)

2) 토픽 파티션 1: 컨슈머 그룹내 컨슈머 4개 => 이럴 경우 리더 컨슈머가 문제가 생길 경우 follower 컨슈머가 그 작업을 대신할 수 있다.

(하지만 컨슈머 스레드 3개는 아무일도 하지 않고 휴면 상태와 같이 있게 된다.) 

 3) 토픽 내 파티션 1개 : 컨슈머 그룹내 컨슈머 2개 => 같은 컨슈머 그룹내에서 동일 토픽에 파티션에는 1개의 컨슈머만접근 가능한 제한으로 인해 한개의 컨슈머는 휴면 상태와 다름 없이 일을 하지 않는다.

4) 토픽내 파티션 4 : 컨슈머 그룹내 컨슈머 2개 => 이럴 경우 컨슈머 1대당 2개의 파티션에 붙어서 R-R방식으로 데이터를 가져와 처리한다.

5) 토픽내 파티션 4 : 컨슈머 그룹내 컨슈머 2개 => 이럴 경우 컨슈머 1대당 2개의 파티션에 붙어서 R-R방식으로 데이터를 가져와 처리한다.

6) 토픽내 파티션 4개 : 컨슈머 그룹 2개 => 각각의 컨슈머 그룹은 토픽에 파티션에 1:1로 맵핑하여 동일한 토픽 메시지를 같이 사용할 수 있다. ( 두 가지 로직을 태울 수 있다.)

위와 같이 컨슈머 그룹이 두 개라면 브로커에 하나의 토픽에 파티션은 consumer group별로 각각의 offset을 가지고 있어서 데이터가 어디까지 처리했는지 구분이 가능하게끔 제공하고 있다.

| 주요 config

• Producer

• consumer

** 리밸런싱(브로커 대상 토픽과 컨슈머의 재맵핑) 과 관련된 주요 설정들

# 요청에 대해 응답을 기다리는 최대 시간

requestTimeoutMs: 30000

# 컨슈머와 브로커사이의 세션 타임 아웃시간.

sessionTimeoutMs: 30000

# session.timeout.ms와 밀접한 관계가 있으며 session.timeout.ms보다 낮아야한다. 일반적으로 1/3

heartbeatIntervalMs: 10000

# 컨슈머가 polling하고 commit 할때까지의 대기시간

maxPollIntervalMs: 600000

# poll ()에 대한 단일 호출에서 반환되는 최대 레코드 수

maxPollRecords: 1

위에서 보면 consumer는 한 번에 하나의 레코드를 가져오고 , polling을 하여 처리대기 시간은 상대적으로 길게 잡았으며, 세션 타임아웃보다 heartbeats 인터벌은 짧게 잡아 컨슈머의 이상여부(정상 실행 상태여부) 를 체크할 수 있게 설정해야 한다.

그리고 세션 타임아웃은 폴링한 데이터가 정상적으로 처리가 가능한 이상 수준의 타임아웃을 가져야 Rebalancing 이슈가 발생하지 않는다.

그러므로 max poll record를 너무 길게 잡으면 세션 타임 아웃 등 설정 정보도 늘어나서 컨슈머가 실제로 정상적인지 체크하는데 어려움이 발생할 수 도 있다.

session.timeout.ms

이 옵션은 heartbeat 없이 얼마나 오랫동안 컨슈머가 있을 수 있는지를 제어하며 heartbeat.interval.ms와 밀접한 관련이 있어서 일반적으로 두 속성이 함께 수정된다.

heartbeat.interval.ms 컨슈머가 얼마나 자주 heartbeat을 보낼지 조정한다. session.timeout.ms보다 작아야 하며 일반적으로 1/3로 설정 ex. 3000 (3초)

max.poll.interval.ms

이러한 경우에 컨슈머가 무한정 해당 파티션을 점유할 수 없도록 주기적으로 poll을 호출하지 않으면 장애라고 판단하고 컨슈머 그룹에서 제외시키도록 하는 옵션이다.300000 (5분)

max.poll.records

이 옵션으로 polling loop에서 데이터 양을 조정 할 수 있다.

enable.auto.commit

백그라운드로 주기적으로 offset을 commit = true

auto.commit.interval.ms

주기적으로 offset을 커밋하는 시간5000 (5초)

auto.offset.reset

이전 offset값을 찾지 못하면 error 발생 Ex.latest

**추가 정보 : kafka vs rabiitMQ vs redis pub/sub 의 비교 

참고 정보 : 

https://velog.io/@hyun6ik/Apache-Kafka-Consumer-Rebalance

https://magpienote.tistory.com/254

https://jyeonth.tistory.com/30

https://velog.io/@holicme7/Apache-Kafka-%EC%B9%B4%ED%94%84%EC%B9%B4%EB%9E%80-%EB%AC%B4%EC%97%87%EC%9D%B8%EA%B0%80

https://www.popit.kr/kafka-consumer-group/