[Kafka] 이론

Intro

서비스가 확장되면 데이터 통합 문제가 훨씬 더 복잡해진다.

정보를 공유하기 위해서는 모든 데이터를 다른 시스템에 전송해야 하기 때문이다.

여기서 데이터의 전반적인 형상이 변하면 어떻게 될까? 모든 연결과 데이터 추출을 위한 요청으로 인해 부하가 증가하게 될 것이다.

여기서 Apache Kafka를 이용해서 스케일링할 수 있다.

Kafka는 분산형이고 회복력 있는 아키텍처이며, 수평적 확장성이 있다.

고성능이며 메세지 처리 규모가 매우 크다.(수백만/s)

 

활용 사례: 메시징 시스템, 활동 추적 시스템, 애플리케이션 로그 수집, 스트림 처리 등

 

---

1️⃣ 토픽, 파티션, 오프셋

• Kafka 토픽
      ➡ Kafka 클러스터 안에 있는 데이터 스트림 즉, 데이터베이스의 테이블(제약조건X)
      ➡  모든 종류의 메시지 형식 지원
      ➡  데이터 스트림: 토픽 안에 있는 메시지들의 순서
• 파티션
      ➡  토픽을 구성하는 단위
      ➡  데이터를 파티션에 기록하면 변경할 수 없다.
      ➡  Kafka 안의 데이터 수정 및 삭제 불가능(불가변성)
• 오프셋
      ➡  파티션 안의 토픽들의 id (JPA IDENTITY 전략의 pk처럼 점점 증가)
      ➡  메시지 순서가 한 파티션 안에서만 보장됨

---

2️⃣ 프로듀서

• 프로듀서는 카프카 토픽과 파티션에 데이터를 전송한다.
      ➡  프로듀서는 어떤 파티션에 어떤 메세지가 기록될 것인지 미리 알고 있다.
• 카프카 서버에서 특정 파티션이 고장날 경우, 어떻게 복구할지 프로듀서가 알게 된다.
• 모든 파티션에 걸쳐 데이터를 전송하여 카프카 내부에서 로드 밸런싱이 일어난다.
• 프로듀서는 메시지 안에 메시지 키를 갖고 있다. (메시지 자체가 데이터를 보유)
      ➡  키: 문자열, 숫자, 바이너리 ... 등
      ➡  예시
      ➡  두 가지 파티션이 있는 토픽에 기록
      ➡  만약 키가 null이면 데이터가 라운드 로빈 방식으로 전송됨
      ➡  (파티션 0 -> 파티션 1 -> 파티션 2 ...) 이런 식으로 로드 밸런싱
  
  
• 동일한 키를 공유하는 모든 메시지들은 해싱 전략 덕분에 항상 동일한 파티션에 기록된다.
      ➡  그래서 키를 지정할 때 특정 필드에 대한 메시지 순서 설정 필요

 

• 카프카 메시지의 모습
  Apache Kafka로 전송되어 저장됨

• 카프카의 기술이 좋은 이유는 프로듀서로부터 입력값으로 직렬로 된 바이트만을 받고,
  출력값으로서 바이트를 컨슈머에게 전송하기 때문
  • 하지만 메시지를 구성할 때에는 바이트가 아님 -> 직렬화 필요
• 메시지 생성 방법: Kafka Message Serializer
      8️⃣  숫자
          ➡  KeySerializer를 IntegerSerializer로 지정
          ➡  카프카 프로듀서가 키 객체를 Seriailzer를 통해 직렬 바이트로 변환 -> 바이너리 표현 나옴
      🆗  문자
          ➡  KeySerializaer를 StringSerializaer로 지정
          ➡  카프카 프로듀서가 키 객체를 Serializaer를 통해 직렬 바이트로 변환

• 이렇게 변환한 메시지를 카프카로 전송할 수 있다.

메시지 키들의 해싱 방법
카프카 파티셔너가 메시지를 받아서 전송할 대상인 파티션 결정
프로듀서 파티셔너 로직이 레코드를 확인하고 파티션 지정
프로듀서가 카프카에 전송
기본 카프카 파티셔너에서 키들이 murmur2 알고리즘을 이용해서 해싱됨

targetPartition = Math.abs(Utils.murmur2(keyBytes)) % (numPartitions - 1)​

 

---

3️⃣ 컨슈머 & 역직렬화

토픽에서 데이터를 읽으려면 컨슈머를 사용해야 한다.

• 컨슈머는 풀(pull) 모델을 구현한다.
• Kafka 브로커(서버)에 데이터를 요청하고 되돌아오는 응답을 받는다.
• 파티션에서 데이터를 읽어야 하는 컨슈머들은 자동으로 카프카 서버에서 읽을지 알게 된다.
• 서버가 고장날 경우 어떻게 복구할지 알게 된다.
• 데이터는 낮은 오프셋부터 높은 오프셋 순서로 읽는다. (파티션끼리는 순서 보장X)

 

이제 컨슈머는 메시지를 읽고 카프카로부터 받은 바이트를 객체나 데이터로 변환해야 한다.

• 바이너리 형식의 키
• 바이너리 형식 및 카프카 메시지에 담긴 값

위의 둘을 읽어서 프로그래밍 언어가 사용할 수 있는 객체로 변환해야 한다.

그러면 컨슈머는 메시지의 형식이 무엇인지 미리 알고 있어야 한다.

 

컨슈머의 역직렬화

 

---

4️⃣ 컨슈머 그룹 & 컨슈머 오프셋

카프카를 사용하고 스케일링을 하려고 할 때 애플리케이션 안에 많은 컨슈머가 있을 것이다.

그러면 그 컨슈머들이 그룹 형태로 데이터를 읽는다. 이것을 컨슈머 그룹이라고 한다.

 

5개의 파티션으로 된 카프카 토픽을 예로 들어보자.

컨슈머 그룹에는 3개의 컨슈머가 있다. 그러면 그 같은 그룹에 속한 각각의 컨슈머는 각각 다른 파티션에서 읽게 될 것이다.

예를 들어 컨슈머 1이 파티션 0과 파티션 1에서 읽는다면, 컨슈머 2는 파티션 2, 3에서 읽고

컨슈머 3은 파티션 4에서 읽는다는 것이다.

이런 식으로 그룹이 카프카 토픽 전체를 읽게 된다.

 

그러면 컨슈머 그룹 안에 있는 컨슈머의 개수가 파티션의 개수보다 많으면 어떻게 될까?

 

3개의 파티션으로 된 카프카 토픽을 예로 들어보자.

이번에는 파티션 개수보다 많은 컨슈머 그룹이 있다.

여기서는 각각의 컨슈머가 하나의 파티션에서 읽고, 남는 컨슈머는 대기 컨슈머가 된다. 어떠한 토픽 파티션에서도 읽지 않게 된다.

 

 

하나의 토픽에 다수의 컨슈머 그룹이 있을 수도 있다.

컨슈머 그룹 안에서는 오직 하나의 컨슈머가 하나의 파티션에 지정될 것이다.

그렇다면 컨슈머 그룹이 있는 이유는 뭘까?

서비스당 하나의 컨슈머 그룹을 갖게 되기 때문이다.

위치 서비스와 알림 서비스를 예로 들면, 한 컨슈머 그룹은 위치 서비스에 사용되고, 다른 컨슈머 그룹은 알림 서비스에 사용되는 것이다.

 

 

컨슈머 그룹 안에서 오프셋을 정의할 수 있다.

카프카는 컨슈머 그룹이 읽고 있던 오프셋, 즉 컨슈머 오프셋을 저장한다.

오프셋이 저장되면 컨슈머는 그 오프셋부터 계속 읽을 수 있게 된다.

 

카프카로부터 받은 데이터에 대한 처리를 컨슈머가 완료하면, 컨슈머는 종종 오프셋을 커밋해야 한다.

그리고 카프카 브로커가 컨슈머 오프셋 토픽에 기록하라고 알린다.

오프셋을 커밋함으로써 카프카 토픽을 어디까지 read를 성공했는지 카프카 브로커에게 알려줄 수 있다.

Why? 컨슈머가 다운되면 다시 돌아와서 읽었던 곳부터 다시 읽을 수 있기 때문

사진을 보면 카프카가 다시 시작할 때 4263부터 읽으라고 전송하는 것을 볼 수 있다.

 

이 컨슈머 그룹 오프셋 덕분에 충돌하거나 실패한 곳부터 데이터를 재생할 수 있는 메커니즘을 갖게 된다.

 

 

컨슈머의 전달 의미론

• 최소한 한 번
      ➡  메시지가 처리된 직후 오프셋이 커밋될 것이라는 의미
      ➡  만약 처리가 잘못되면 메시지를 다시 읽을 기회가 주어짐
• 최대한 한 번
      ➡  컨슈머가 메시지를 받자마자 오프셋 커밋
      ➡  처리가 잘못되면 메시지를 잃게 됨(메시지를 실제로 처리하기 전에 오프셋을 커밋해서)
• 정확히 한 번
      ➡  메시지를 딱 한 번만 처리

---

5️⃣ 브로커 & 토픽

카프카 클러스터는 다수의 카프카 브로커들로 구성되어 있다.

브로커는 ID로 식별한다.

 

파티션이 3개인 토픽A, 파티션이 2개인 토픽B를 예로 들어보자.

그리고 3개의 브로커 101, 102, 103가 있다고 해보자.

데이터와 파티션이 모든 브로커에 걸쳐 분산되었다. 이를 수평적 스케일링이라고 한다.

여기서 볼 건 브로커가 모든 데이터를 갖는 게 아니고, 가져야 할 데이터만 갖는다는 것이다.

 

부트스트랩 서버

    ➡  클러스터에 있는 각각의 카프카 브로커

 

클러스터에서 브로커 하나에만 연결하면 클라이언트는 전체 클러스터에 연결하는 방법을 알게 된다.

만약 브로커 101에 대한 연결에 성공하면, 브로커 101은 클러스터 안에 있는 모든 브로커의 리스트를 리턴하게 된다.

그러면 카프카 클라이언트는 브로커 리스트 덕분에 필요한 브로커에 연결할 수 있게 된다.

 

---

6️⃣ 토픽 복제

 

보통 실제 카프카 클러스터가 있을 때 복제 계수를 1보다 큰 숫자로 설정해야 한다.

보통 2와 3 사이로 설정하고, 3으로 설정하는 경우가 가장 많다.

이렇게 하면 만약 브로커가 다운될 경우 다른 카프카 브로커에 데이터 사본이 있어서 이를 주고 받을 수 있게 된다.

 

아래를 예로 들어보자.

토픽 A와 파티션이 2개가 있고, 복제 계수가 2이다. 그리고 3개의 카프카 브로커가 있다.

복제 계수가 2라면 복제 매커니즘에 의해 파티션 0의 사본이 브로커 103에 있게 된다.

만약 여기서 브로커 102가 다운되면 어떻게 될까?

사진대로 브로커 101, 103이 업 상태이므로 여전히 데이터를 제공할 수 있다.

간단히 말해서 브로커 102가 다운돼도 괜찮게 된다.

 

 

복제를 하게 되면 파티션의 리더가 존재하게 된다. 오직 1개의 브로커만 파티션의 리더가 될 수 있다.

그리고 규칙에 따르면, 프로듀서는 파티션의 리더인 브로커에게만 데이터를 전송할 수 있다.

별 표시가 파티션의 리더

만약 데이터가 잘 복제되면 데이터 복제의 동기화가 이루어진다.

리더의 기본적인 작동 방식에 따르면 프로듀서는 어떤 파티션의 리더 브로커에만 기록하게 된다.

그래서 만약 프로듀서가 파티션 0으로 전송하길 원한다는 걸 알고, 우리가 리더와 ISR(In-Sync-Replica)이 있다면, 프로듀서는 파티션의 리더인 브로커에게만 데이터를 전송해야 한다는 걸 알고 있다.

카프카 컨슈머는 기본 값으로서, 어떤 파티션의 리더로부터만 읽게 된다.

 

카프카 컨슈머 레플리카 페칭

컨슈머가 가장 가까운 레플리카에서 읽게 해주는 기능

만약 파티션 0의 리더인 브로커 101이 프로듀서로부터 데이터를 받고, 데이터를 브로커 102의 ISR 파티션 0에 복제했다고 가정하자.

그러면 컨슈머는 레플리카 자체에서 읽을 수 있게 된다.

이는 컨슈머가 브로커 102에 아주 가까이 있을 수 있다면 브로커 102를 읽을 수 있는 것이다.

클라우드를 사용한다면 네트워크 비용을 줄일 수 있는 기능이다.

 

---

7️⃣ 프로듀서 확인 & 토픽 내구성

 프로듀서는 데이터 쓰기가 성공적으로 이루어졌다는 확인을 카프카 브로커로부터 받을 수 있다.

• acks=0
      ➡  프로듀서가 확인을 기다리거나 요청하지 않는다.
      ➡  즉, 데이터가 유실될 수도 있다.
• acks=1
      ➡  프로듀서가 리더 브로커가 확인하기를 기다린다. 그래서 데이터 유실이 제한된다.
• acks=all
      ➡  모든 레플리카를 리더가 쓰기를 확인하라고 요구한다.
      ➡  어떤 상황에서도 데이터가 유실되지 않는다는 걸 보장한다.

 

만약 복제 계수가 3인 카프카 토픽과 토픽이 브로커 2개의 손실을 견딜 수 있다고 가정해보자.

 

만약 복제 계수가 2이고 브로커가 3개라면 브로커 102를 잃었을 때에도 여전히 토픽 데이터가 제공된다.

왜냐하면 다른 브로커에 있기 때문이다.

그래서 N의 복제 계수를 가지면 영구적으로 최대 N-1개의 브로커를 잃어도 된다.

---

정리

✅ 카프카 클러스터는 다수의 브로커로 구성된다.

✅ 클러스터 안에서 토픽, 파티션, 레플리카 파티션 리더, ISR이 존재한다.

✅ 프로듀서는 소스 시스템에서 원하는 데이터를 받고, 데이터를 카프카로 전송한다.

✅ 컨슈머는 서버에 데이터를 요청하고 응답을 받는다.

✅ 라운드 로빈: 데이터가 토픽 안의 모든 파티션에 걸쳐 분산된다.

✅ 키를 지정하면 동일한 키가 결국 동일한 파티션으로 간다.

✅ acks=0, 1, all

✅ 컨슈머 오프셋을 어떻게 저장하는지?: 최소한 한 번, 최대한 한 번, 정확히 한 번

✅ 카프카 클러스터를 주키퍼로 관리한다.

✅ 주키퍼에는 리더와 팔로워 개념이 있고, 브로커 및 메타데이터 관리를 한다.

✅ 주키퍼에서 KRaft 모드로 카프카 클러스터만 사용하는 방향으로 전환하고 있다.(2020년 ~)