카프카 핵심가이드 리뷰

Consumer

Consumer group

• 브로커는 성능을 위해 하나의 토픽을 여러 파티션으로 병렬 구성하여 처리합니다.
• 하지만 둘 이상의 파티션을 하나의 컨슈머로만 처리한다면 성능 상의 문제가 발생할 수 있습니다.
• 그래서 카프카 컨슈머는 하나 이상의 컨슈머가 컨슈머 그룹(Consumer Group)을 구성하여 하나의 토픽을 구독할 수 있습니다.

  

  
  출처: https://www.oreilly.com/library/view/kafka-the-definitive/9781491936153/ch04.html#T1_four_partitions

kafka 내부 mechanism

클러스터 멤버쉽

• 각 브로커는 브로커 설정 파일에 정의되었거나 자동으로 생성된 고유한 식별자를 가진다.
• 브로커 프로세스는 시작될 때 주키퍼에 Ephemeral 노드의 형태로 ID를 등록하고, 카프카 브로커들과 몇몇 생태계 툴들은 해당 ID가 저장된 /brokers/ids를 구독하여 이벤트를 받는다.

컨트롤러

• 일반적인 카프카 브로커의 기능 + 파티션 리더를 선출하는 역할
• 클러스터에서 가장 먼저 시작되는 브로커는 주키퍼의 /controller 에 Ephemeral 노드를 생성함으로써 컨트롤러가 된다.
• 다른 브로커는 '노드가 이미 존재함' 예외를 받으면 컨트롤러 노드가 이미 있다는 걸 알아차리게 된다.
• 브로커들은 주키퍼의 컨트롤러 노드에 변동이 생겼을 때 알림을 받기 위해 '와치'를 설정하고, 클러스터 안에 한 번에 한 개의 컨트롤러만 있도록 보장한다.

Ephemeral 노드가 삭제될 경우

• 와치를 통해 이를 알게 되고, 주키퍼에 컨트롤러 노드를 생성하려고 시도한다.
• 가장 먼저 새로운 노드를 생성한 브로커가 다음 컨트롤러가 되고, 다른 브로커는 새 컨트롤러 노드에 대한 와치를 다시 생성한다.
• 컨트롤러 좀비
  • 새로운 컨트롤러가 선출될 때마다 '증가된 에포크 값'을 전달받는다.
  • 이것은 컨트롤러 브로커가 오랫동안 가비지 수집 때문에 멈춘 사이 주키퍼 사이의 연결이 끊어질 수 있기 때문에 중요하다.
  • 컨트롤러가 전송하는 메시지에 컨트롤러 에포크를 포함하면 브로커는 예전 컨트롤러가 보내온 메시지를 무시할 수 있다.
  • 브로커가 컨트롤러가 되면 클러스터 메타데이터 관리와 리더 선출을 시작하기 전에 먼저 주키퍼로부터 최신 레플리카 상태 맵을 읽어온다.
  • 브로커가 클러스터를 나가면, 컨트롤러는 해당 브로커가 리더를 맡고 있던 모든 파티션에 대해 새로운 브로커를 할당해준다.

KRaft 모드

• 주키퍼 기반 컨트롤러로부터 탈피해서, 래프트 기반 컨트롤러로 옮겨간다.
• 왜 카프카 커뮤니티는 컨트롤러를 교체하기로 결정했을까?
  • 컨트롤러가 주키퍼에 메타데이터를 쓰는 작업은 동기적으로 이루어지지만, 브로커 메시지를 보내는 작업은 비동기적으로 이루어진다. 그렇기 때문에 브로커, 컨트롤러, 주키퍼 간에 메타 데이터 불일치가 발생할 수도 있다.
  • 컨트롤러가 재시작될 때마다 주키퍼로부터 모든 브로커와 파티션에 대한 메타데이터를 읽어오고, 그것들을 모든 브로커로 전송해야 하는데 브로커와 파티션 수가 증가함에 따라 컨트롤러 재시작은 느려진다.

멱등적(idempotent) 프로듀서

• 멱등적이다 == 동일한 작업 실행해도 한 번만 실행한 것과 결과가 같다. (프로듀서 재시도로 인한 중복 방지)

작동 원리

• 멱등적 프로듀서 기능을 켜면 모든 메시지는 고유한 프로듀서 ID(PID)와 sequence number를 가지게 된다.
• 각 메시지의 고유한 식별자: 대상 토픽 및 파티션 + PID + Sequence number
• 각 브로커는 해당 브로커에 할당한 모든 파티션들에 쓰여진 마지막 5개 메시지들을 추적하기 위해 이 고유 식별자를 사용
• max.in.flights.requests.per.connection(def=5,max=5: 파티션별로 추적되어야 하는 시퀀스 넘버의 수를 제한하고 싶다면 해당 설정을 5 이하로 잡으면 된다.

작동 실패했을 경우

1. 프로듀서 재시작

• 프로듀서에 장애가 발생할 경우, 보통 새 프로듀서를 생성해서 장애가 난 프로듀서를 대체한다.
• 멱등적 프로듀서 기능이 켜있는 경우, 프로듀서는 프로듀서가 시작될 때 초기화 과정에서 카프카 브로커로부터 프로듀서 ID를 생성받는다.
• 기존 프로듀서가 작동을 멈췄다가 새 프로듀서가 투입된 이후 작동을 재개해도, 서로 다른 PID를 가졌기 때문에 기존 프로듀서는 좀비로 취급되지 않는다.

2. 브로커 장애

• 브로커에 장애가 발생할 경우, 컨트롤러는 장애가 난 브로커가 리더를 맡고 있는 파티션에 대해 새 리더를 선출하게 된다.
• 리더는 새 메시지가 쓰여질 때마다 인-메모리 프로듀서 상태에 저장된 최근 5개의 시퀀스 넘버를 업데이트한다.
• 팔로워 레플리카는 리더로부터 새로운 메시지를 복제할 때마다 자체적인 인-메모리 버퍼를 업데이트한다.
• 즉, 팔로워가 리더가 되어도 이 버퍼를 활용해 중복처리를 진행한다.
• 만약 예전 리더가 복구된다면, 재시작 후에는 인-메모리 프로듀서 상태는 더 이상 메모리에 저장되어 있지 않다.
• 그래서 복구 과정에 도움이 되도록 브로커는 종료되거나 새 세그먼트가 생성될 때마다 프로듀서 상태에 대한 스냅샷을 파일형태로 저장한다.

Exactly Once Semantics (EOS) 구현하기

• 중복 없이 정확히 한 번만 처리를 보장

Transactional Producer 설정

producer = Producer({
    'bootstrap.servers': brokers,
    'transactional.id': 'eos-transactions.py'
})
producer.init_transactions()
producer.begin_transaction()

• transactional.id: 프로듀서를 트랜잭션 모드로 동작
• 읽고 쓰는 모든 행동을 하나의 묶음으로 만든다.

메시지 읽고 변환 (Consume and Transform)

msg = consumer.poll(timeout=1.0)
processed_key, processed_value = process_input(msg)

• consumer.poll()로 메시지 읽어오기
• process_input(msg)는 읽은 메시지의 key와 value를 Base64로 인코딩해서 변환

메시지 쓰기

producer.produce(output_topic, processed_value, processed_key,
                 on_delivery=delivery_report)

• 변환한 메시지를 새로운 토픽(output_topic)에 보냄.

Consumer offset과 함께 커밋

• 매 100개의 메시지를 처리할 때마다 트랜잭션을 커밋

  if msg_cnt % 100 == 0:
    producer.send_offsets_to_transaction(
        consumer.position(consumer.assignment()),
        consumer.consumer_group_metadata())
    producer.commit_transaction()
    producer.begin_transaction()

• send_offsets_to_transaction: 현재 읽은 위치(offset)를 트랜잭션에 저장

• commit_transaction: 읽은 메시지들과 쓴 메시지들, offset을 한꺼번에 커밋.

• begin_transaction: 새 트랜잭션 시작.