주키퍼, 브로커, 프로듀서, 컨슈머

주키퍼

• 카프카 클러스터 관리 용도로만 간략히 알고 넘어가자
• 기존에는 카프카 사용 시 주키퍼가 필수였으나, 주키퍼 사용시 제약사항이나 한계성에 의해 주키퍼와 카프카의 연결고리를 끊고 있다.

Apache Kafka 2.8 버전부터 주키퍼 대신 kraft 를 사용할 수 있으나, 2.8.x 버전 부터 3.2.x 버전까지는 개발 단계(얼리 액세스,early access) 입니다. Kafka 3.3.1 버전에서 부터 Production(운영)에서 사용할 준비가 되었음 발표하면서 KRaft 지원과 ZooKeeper 지원종료(제거)에 대한 릴리즈 타임라인을 발표 하였습니다.

 

브로커

• 일반적으로 카프카라고 불리는 시스템(어플리케이션)을 말한다.
• 컨트롤러
  • 다수의 브로커 중 한대가 컨트롤러 역할을 한다
    • 파티션 할당, 정상 동작 모니터링
    • 브로커의 상태를 체크하고, 브로커가 클러스터에서 빠지는 경우 해당 브로커에 존재하는 리더 파티션을 재분배하는 역할을 한다.
• 그룹 코디네이터
  • 다수의 브로커 중 한대가 코디네이터 역할을 한다
    • 컨슈머 그룹의 상태를 체크하고 파티션을 컨슈머와 매칭되도록 분배하는 역할
    • 컨슈머를 재할당하는 과정을 리밸런스라고 한다.
• 메시지 저장과 메세지 파일 관리
  • 브로커는 프로듀서로부터 전달되는 메세지를 로그형태로 log.dir에 설정한 경로에 저장한다.
  • 각 브로커에 저장되는 메세지는 파티션 별로 세그먼트라는 파일로 저장된다.
• 데이터 삭제
  • 로그 세그먼트 단위로 삭제한다
  • Delete 옵션에 의해 일정 시간이나 용량에 따라 데이터를 삭제한다
    • 액티브가 아닌 세그먼트를 삭제한다.
      • 레코드단위 개별 삭제는 불가능하다.
    • 보통 3일정도로 유지하는 것이 일반적 (금, 토, 일)
    • 브로커가 세그먼트가 삭제 영역에 들어왔는지 계속 확인해서 삭제한다.
  • Compact 옵션으로 가장 최신의 메세지키를 제외한 데이터를 삭제할 수도 있다.
    • 액티브 세그먼트를 제외한 나머지 세그먼트에 한해서 이루어진다.
    • 클린 로그 : 압축이 완료된 로그
    • 더티 로그 : 압축이 완료되지 않은 로그(중복된 메세지 키가 있다)
    • 클린레코드의 개수와 더티레코드의 개수에 따라 압축시작을 수행하게 된다.
      • 4:4 → 50% / 50%가 도달 했을 때 압축하겠다 이런식으로 설정 가능
• 복제
  • 파티션 단위로 데이터를 복제한다.
  • 토픽 생성할 때 파티션의 복제 개수를 설정한다. (설정 안하면 브로커의 기본 설정)
  • 최소값 1 (복제 없음) ~ 최대값 (브로커의 개수만큼)
    • 보통 2-3정도로 설정한다.
  • 리더 파티션 : 프로듀서, 컨슈머랑 직접적인 통신을 한다
  • 팔로워 파티션 : 데이터를 복제만 한다.

 

프로듀서

• 이벤트를 발행하는 클라이언트 어플리케이션
• 프로듀서는 라우팅 계층없이 파티션의 리더인 브로커에 데이터를 직접 전송한다.
  • 이렇게 할 수 있기 위해서는 모든 카프카 노드는 언제든지 어떤 서버가 활성 상태이고, 토픽 파티션의 리더는 누구인지에 대한 메타데이터 요청에 응답할 수 있어야 한다.
    • 메타데이터
      • 카프카 클라이언트는 통신하고자하는 리더 파티션의 위치를 알기 위해 메타데이터를 브로커로부터 전달 받는다
      • 메타 데이터 캐시 유지하는 기간 설정 가능
      • 메타 데이터에 이슈가 발생한다면? 잘못된 브로커로 요청 → leader_not_avaliable 익셉션 발생
• Partitioner
  • 어느 파티션으로 토픽을 전송할지 설정한다.
  • 컨슈머가 지역성이 중요한 데이터를 처리할 수 있게 해준다.
  • 파티션 개수가 변경되는 순간, 메세지 키와 파티션 번호 매칭이 깨지게 됨. 처음부터 파티션 그냥 충분히 큰 개수로 운영해라
  • UniformStickyPartitioner : 기본값
    • 메세지 키가 있는경우 : 해시값과 파티션 매칭
    • 메세지 키가 없는경우 : 라운드로빈 효율 극대화
      • 배치로 묶일때까지 기다렸다가 전송한다
  • RoundRobinPartitioner
    • 메세지 키가 있는경우 : 해시값과 파티션 매칭
    • 메세지 키가 없는경우 : 파티션 순회
  • 라운드로빈 or 의미 부여해서 파티셔닝 등등
    • 직접 구현도 가능
• 옵션
  • bootstrap.servers : 연결할 서버 정보
  • key.serializer, value.serializer : 직렬화, 역직렬화 클래스
  • partitioner.class : 파티셔너 클래스
  • retries : 에러 났을 때 재시도할 횟수
  • acks : 브로커들에 정상적으로 저장되었는지 전송 성공 여부 확인
    • 데이터를 프로듀싱하고 래플리카가 승인할때까지 기다린다
      • 0 : 데이터 전송할 때, 리더파티션에 데이터 저장되었는지 확인하지 않겠다. 데이터 유실 괜찮을 때 사용
      • 1 (리더에만 잘 적재되면 성공) : 기본값
      • -1 : min.insync.replicas 옵션값에 지정된 개수만큼 확인
• send : Future 객체를 반환한다.
  • 동기, 비동기로 결과 확인 가능

 

KafkaProducer Client

• KafkaProducer : send()를 호출함으로써 Record를 전송한다.
  • 사용자는 send() 호출 시 전송할 Record와 전송 완료 후 실행할 콜백을 지정할 수 있다. send()가 호출되면 Serialization, Partitioning, Compression 작업이 이루어지고 최종적으로 RecordAccumulator에 Record가 저장된다.
• RecordAccumulator
  • 사용자가 KafkaProducer의 send()를 호출하면 Record가 바로 Broker로 전송되는 것이 아니라 RecordAccumulator에 저장된다. 그리고 실제로 Broker에 전송되는 것은 이후에 비동기적으로 이루어진다.
    • RecordAccumulator는 batches라는 Map을 가지고 있는데
      • 이 Map의 Key는 TopicPartition이고, Value는 Deque<RecordBatch>이다.
      • RecordAccumulator의 append()가 호출되면 batches에서 추가될 Record에 해당하는 TopicPartition의 Deque를 찾는다. 이 Deque의 Last에서 RecordBatch 하나를 꺼내서 Record를 저장할 공간이 있는지 확인한다. 여유 공간이 있으면 해당 RecordBatch에 Record를 추가하고, 여유 공간이 없으면 새로운 RecordBatch를 생성해서 Last쪽으로 저장한다.
• Sender : KafkaProducer는 별도의 Sender Thread를 생성한다.
  • Sender Thread는 RecordAccumulator에 저장된 Record들을 Broker로 전송하는 역할을 한다. 그리고 Broker의 응답을 받아서 사용자가 Record 전송 시 설정한 콜백이 있으면 실행하고, Broker로부터 받은 응답 결과를 Future를 통해서 사용자에게 전달한다.
  • Sender Thread는 RecordAccumulator에 저장된 Record를 꺼내서 Broker로 전송하고 Broker의 응답을 처리한다.

프로듀서 메세지 전달 보장

• At most once : 최대 한번 전송. ACK를 기다리지 않는다.
• At least once : 적어도 한번 전송
  • ACK를 보냈는데 그 ACK가 유실된 경우 데이터 중복 전송됨
• Exactly once : 딱 한번 전송. 멱등성 옵션

프로듀서 멱등성 옵션

• enable.idempotence
• 브로커는 각 프로듀서에 ID를 할당하고, 프로듀서는 모든 메세지를 시퀀스 번호와 함께 전송한다.
  • 프로듀서 PID와 시퀀스 넘버를 함께 전달한다
• 브로커는 이 번호를 사용해 중복 메세지를 제거한다 -> 동일한 데이터를 여러번 전송하더라도, 카프카 클러스터에 단 한번만 저장되는것을 의미한다
• 한계
  • 동일한 세션에서만 정확히 한번 전송을 보장한다 -> 프로듀서가 다시 뜬다면 프로듀서 PID가 바뀌기 때문에 보장 안됨.

트랜잭션 프로듀서와 트랜잭션 컨슈머

• 모든 메세지를 기록하는데 성공하거나, 그게 아니면 아무것도 기록하지 않는다.
• 트랜잭션 프로듀서가 커밋하면 트랜잭션 컨슈머가 가져감
• 트랜잭션 컨슈머
  • isolation level : read_commited (커밋된 레코드만 읽어라)

 

컨슈머

• 이벤트를 구독하는 어플리케이션
• 커밋 : 데이터를 어디까지 읽었는지 표시한다.
  • poll()은 이전에 커밋한 오프셋이 있으면 그 오프셋 이후의 오프셋을 읽어온다
  • 커밋이 정상적으로 처리되지 않았을 경우, 데이터 처리에 중복이 발생할 가능성도 있음
    • 컨슈머의 재시작이 이루어질 경우를 생각
  • 자동커밋/수동커밋
    • 자동커밋 : enable.auto.commit
      • Consumer에서 poll()을 호출할 때 auto.commit.interval.ms(default: 5s)이 지났는지를 확인 한 뒤 Commit이 가능 하면 가장 마지막 offset을 commit한다.
      • 중복 가능성 : 커밋 하기전에 리밸런싱 발생 시
      • 손실 가능성 : 커밋은 되었지만 메세지 처리가 끝나지 않았는데 장애 발생한 상황
    • 수동커밋
      • 동기커밋
      • 비동기커밋
• poll을 호출하기전에 내부적으론 데이터를 가져와 있다.
• 컨슈머에서 데이터 처리 시 Exception이 발생하는 사례들
  • 메시지를 deserialize 할 수 없는 경우
  • 잘못된 메시지가 들어온 경우(데이터의 형식이 안 맞는다든지, 허용 범위에서 벗어나 valid하지 않은 값이 들어온다는지)
  • 컨슈머의 비즈니스 로직을 잘못 짠 경우
  • 처리되지 않는 메세지를 모으는 큐가 별도로 존재해 다음과 같이 핸들링 하는 방법도 있다.
    • DLQs : factory.setErrorHandler(new SeekToCurrentErrorHandler( new DeadLetterPublishingRecoverer(kafkaProducerTemplate), 3)); // 3번 Retry -> DLT
• 옵션
  • bootstrap.servers : 연결할 서버 정보
  • fetch.min.bytes : 한번에 가져오는 최소 사이즈
  • group.id : 컨슈머 그룹 아이디
  • auto.offset.reset : 커밋 한번도 안했을 때 어느 오프셋부터 읽을지 선택하는 옵션
    • 새로만든 컨슈머 운용할 때 사용됨.
  • enable.auto.commit : 자동 커밋 유무
  • heartbeat.interval.ms : 그룹 코디네이터에게 보내는 하트비트 간격
  • session.timeout.ms : 해당 주기마다 컨슈머가 종료된 것인지 판단한다. 이 시간 전까지 하트비트를 보내지 않으면 컨슈머는 종료된 것으로 간주하고 컨슈머그룹에서 제외하고 리밸런싱을 시작
• 안정적으로 종료
  • WakeUpException 발생 시켜서 종료

컨슈머 렉

• 프로듀서가 마지막으로 넣은 오프셋 - 컨슈머가 마지막으로 읽은 오프셋 : 컨슈머렉
• 컨슈머가 읽어들이는 속도 < 프로듀싱 되는 속도
• 해소 방법 : 파티션 개수를 늘리고 컨슈머 개수도 늘린다.
• 모니터링 툴 : Burrow, AKHQ.IO, Kafdrop, Kafka-ui

KafkaConsumer Client

• KafkaConsumer는 ConsumerNetworkClient, SubscriptionState, ConsumerCoordinator, Fetcher, HeartBeat 스레드로 구성된다.
• ConsumerNetworkClient는 KafkaConsumer의 모든 네트워크 통신을 담당한다.
• KafkaConsumer의 모든 요청은 ConsumerNetworkClient의 send 메서드를 통해 시작된다. ConsumerNetworkClient는 send 메서드를 통해 전달된 모든 요청을 ClientRequest로 바꾼다.
  • ClientRequest에는 요청이 완료되었을 때 호출될 RequestFuture가 설정되어 있다.
  • 이 RequestFuture는 ConsumerNetworkClient의 send 메서드를 호출한 콜러(caller)에게 반환된다. 콜러는 RequestFuture를 통해 비동기 요청이 완료되었는지 확인한다.
• ConsumerNetworkClient는 ClientRequest를 바로 전송하지 않고 내부 버퍼인 Unsent Map에 먼저 저장한다. Unsent Map의 Key는 요청을 전송할 브로커의 호스트이고 Value는 브로커로 전송해야 하는 ClientRequest의 리스트이다.
• 컨슈머 리밸런스
  • 리밸런스가 발생한 컨슈머 그룹은 리밸런스 작업이 끝날 때까지 브로커로부터 데이터를 가져오지 않는다. 따라서 리밸런스가 진행되는 동안에는 컨슈머들이 일시 정지된 것처럼 보인다
• Process 스레드가 정상적으로 동작하지 않는다면 max.poll.interval.ms으로 감지가 된다. 만약 KafkaConsumer가 정상이 아닌 경우에는 session.timeout.ms로 감지된다
• Fetcher
  • fetchedRecords 메서드는 내부 캐시인 nextInLineRecords와 completedFetches를 확인하여 브로커로부터 이미 가져온 데이터가 있는 경우에는 max.poll.records 설정 값만큼 레코드를 반환한다

컨슈머 그룹

• 목적에 따라 데이터를 처리하는 컨슈머들을 묶은 그룹. 그냥 같은 로직을 가지고 있고, 스케일아웃한 컨슈머들이라고 생각하면 편하다.
• 파티션 : 컨슈머 = 1 : 1이 최고 효율이다.
  • 파티션 개수보다 컨슈머 그룹이 많아지게 되면, 컨슈머는 놀게 된다
    • 파티션 2개/한 그룹안에 컨슈머 3개
      • 파티션당 컨슈머는 하나여야 하기 때문에
    • 처리량이 떨어져서 컨슈머를 늘려야 한다면 파티션도 늘려야 한다.
• 두개 이상의 컨슈머 그룹을 생성하는 이유?
  • elk에도 로그 적재, hadoop에도 로그 적재 등 같은 데이터를 여러 소스에서 사용해야 하는 경우
• 한개 파티션은 컨슈머 그룹의 한 개 컨슈머에만 연결이 가능하다
  • 그룹에 속해있는 컨슈머들이 파티션 공유 불가능
  • 컨슈머 그룹 기준으로 파티션의 메세지가 순서대로 처리된다.
  • 한개 파티션을 서로 다른 그룹의 컨슈머는 공유할 수 있다
• 컨슈머 그룹을 통해서 Kafka는 가용성 확보와 병렬 처리를 한다. 만약 컨슈머 그룹 내에 특정 컨슈머의 처리가 일정 시간(max.poll.interval.ms 설정만큼) 정지된다면 해당 컨슈머는 그룹에서 제외되고 나머지 컨슈머들로만 데이터가 분배된다.

Assignor

• 컨슈머와 파티션의 할당 정책
• 컨슈머 파티션 할당 전략은 프로듀서의 파티셔너가 어떤 파티션으로 메시지를 전송할 지 결정했던 것처럼 컨슈머가 어떤 파티션의 메시지를 읽어올지를 결정하는 정책
• RangeAssignor : 파티션을 숫자로 정렬. 컨슈머를 사전 순서로 정렬 - 기본값
• RoundRobinAssignor : 모든 파티션을 컨슈머에서 번갈아가면서 할당
• StickyAssignor : 최대한 파티션을 균등하게 배분하면서 할당
  • 첫 번째 목적은 가능한 균형 잡힌 파티션을 할당하는 것
  • 두 번째 목적은 재할당이 발생할 때 되도록 기존의 할당된 파티션 정보를 보장하는 것

 

레플리카

• 고가용성을 위한 개념
• 복제수만큼 파티션의 복제본이 각 브로커에 생긴다
• 리더 : 하나. 리더를 통해서만 메세지를 처리한다
• 팔로워 : 나머지. 리더로부터 데이터를 복제한다.

ISR

• In Sync Replicas
• 리더 파티션과 팔로우 파티션이 모두 Sync가 된 상태
  • Sync : 커밋된 오프셋이 같다
• ISR이 안되었을 때 리더파티션에 장애가 터진다면?
  • 싱크 안된 상태에서 새로운 리더 선출되면 싱크 안된 데이터 유실 가능성 있음!
  • unclean.leader.election.enable = true ← 유실 감수
  • false면 유실 감수하지 않음. 해당 브로커가 복구될 때까지 중단!

스태틱 멤버십

• group.instance.id
• session.timeout.ms 을 큰 값으로 지정해야한다. 이 시간동안 하트비트를 받지 못하다면, 강제로 리밸런싱 일어남.
• 스태틱 멤버십은 컨슈머 그룹 내에서 컨슈머가 재시작 등으로 그룹에서 나갔다가 다시 합류하더라도 리밸런싱이 일어나지 않게 한다
• 스태틱 멤버십이 적용된 컨슈머의 경우,
  • 컨슈머 그룹에서 떠날 때 그룹 코디네이터에게 알리지 않음
  • 다시 컨슈머 그룹에 합류 시에도 그룹 코디네이터는 ID 확인 후 기존 구성원임을 인지

 

 

https://tillog.netlify.app/posts/kafka-consumer-exception