> For the complete documentation index, see [llms.txt](https://ssosso.gitbook.io/study/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://ssosso.gitbook.io/study/messagequeue/kafka/2.md).

# 2. 카프카 기초

## 카프카 기초

### 카프카 5가지 구성요소

1. 브로커 : 메시지 전달, 수신
2. 프로듀셔 : 메시지 생산자, 브로커에 메시지 전송하는 애플리케이션
3. 컨슈머 : 메시지를 수신, 취득하여 소비하는 애플리케이션 
4. 메시지 : 카프카에서 데이터의 최소 단위. 로그한줄 한줄, 센서데이터 등. Key Value를 갖음. 메시지 전송시 파티셔닝 에 이용
5. 토픽 : 메시지를 종류별(토픽,그룹)로 관리하는 스토리지 같은 역할. `브로커에 배치`되어 관리.

### 메시지 송수신 기본

5개의 구성요소로 송수신 구조

> **프로듀셔 -> 메시지 -> 브로커(토픽) -> 메시지 -> 컨슈머**

### 시스템 구성

#### 브로커

* 하나의 서버 or 인스턴스 당 하나의 데몬 프로세스로 동작  
* 메시지 수신/전달 처리
* 여러 대의 클러스터로 구성가능. 브로커(리소스)를 추가 함으로써 스케일아웃
* 브로커가 받은 메시지 -> 디스크에 영속화가 됨 -> 스토리지 용량에 따라 증설검토. 장기간 보존 가능
* ex) `A시스템의 시스템로그`, `웹서비스 B의 엑세스로그` 단위로 `토픽`생성, 데이터 수신/전달

#### Producer API / Consumer API

* **브로커**로 `메시지 전달 및 수신`
* 각각 API를 통해 처리. Java로 제공됨. `브로커 처럼 데몬프로세스로 작동하는 것이 아님`

#### Producer

* 프로듀서 API를 이용하여 `브로커에 데이터 전달`하는 `애플리케이션`
* 각종 로그 전송 및 미들웨어와 연동하여 작동
* **Apache Log4j(Kafka Appender)**
  * 로그 출력시 사용
  * Kafka Appender
* **Apache Flume**
  * 다량의 로그데이터를 효율적으로 수집, 취합, 이동하기 위한 분산형 SW
  * Kafka Sink
* **Fluentd**
  * 크로스 플랫폼 오픈소스 데이터 수집 SW
  * flunt-plugin-kafka
* **Logstash**
  * 엘라스틱 제공 로그수집 엔진
  * logtash-output-kafka

#### Consumer

* 브로커로 부터 `메시지 취득,수신`
* 디스크에 영속화 -> 메시지 취득시 제약이 없어야함
* 디스크에 영속화 되어 있는 동안 메시지 취득 가능
* 데이터 추출 및 실시간 처리를 위한 애플리케이션 데이터 입력등으로 이용됨
* **Apache Spark**
  * 빅데이터 처리를 위한 OSS, 클러스터 컴퓨팅 프레임워크 
  * Spark Streaming + Kafka Integraion Guide
* **Apache Samza**
  * 스트림 처리용 OSS, 준 리얼타임 비동기 계산프레임워크
* **Apache Flink**
  * 스트림 처리용 OSS
* **Apache Flume**
  * Kafka Source
* **Fluentd**
  * fluent-plugin-kafka
* **Logstash**
  * logstash-input-kafka

#### 카프카 클라이언트

토픽 작성 등 카프카의 동작 및 운영 상에 필요한 조작 실행 서버. `메시지의 송수신 처리 서버는 아님`

#### 카프카 클러스터

* 카프카는 여러 대의 브로커 서버, 주키퍼 서버로 이루어진 `클러스터링 메시지 중계 기능`과 `메시지 송수신을 위한 라이브러리 그룹으로 구성`

### 분산 메시징 구조

#### 파티션

* 브로커 데이터를 읽고 쓰는 단위로 분할
* 토픽을 구성하는 파티션 -> 클러스터안에 분산 배치
* 분산배치 된 토픽내의 파티션은 프로듀서로 부터 메시지를 수신 -> 컨슈머에 분산해서 전달
* `하나의 토픽`에 대해 `여러 파티션을 두어 분산 처리함`

#### 컨슈머 그룹

* 카프카는 하류시스템(컨슈머)에 분산스트림 처리 고려 설계 됨
* 여러 컨슈머가 단일 토픽이나 여러 파티션에서 메시지를 취득 -> `컨슈머그룹화`
* 카프카 클러스터 -> `글로벌 ID`를 컨슈머 그룹에 전체에 공유, `여러 컨슈머는 자신이 속한 컨슈머 그룹 식별`
* `식별 값 -> 파티션 분류 및 재시도 제어`

#### 오프셋

* **파티션단위로 메시지의 위치를 나타내는 개념**
* 컨슈머가 취득하는 메시지의 범위 및 재시도 제어
* `Log-End-Offset(LEO)` : 파티션 데이터 끝 표현, `파티션 관한 정보 관리 및 업데이트`
* `Current Offset` : 컨슈머가 어디까지 메시지를 읽었는가 표현, `컨슈머 그룹마다 보관되어 관리 업데이트`
* `Commit Offset` : 컨슈머가 어디까지 커밋했는지 표현, `컨슈머에서의 데이터 취득을 계기로 업데이트`

### 메시지 송수신

#### 프로듀서의 메시지 송신

버퍼기능 : 메모리에 쌓아둔 후 벌크처리 1. 지정한 크기를 축적하여 전송 (버퍼) 2. 지정한 대기시간에 도달시 전송 (버퍼)

#### 컨슈머의 메시지 취득

취득대상 -> `토픽`과 `파티션(Current Offset)`의 마지막 취득 메시지 부터 브로커에서 보관하는 최신메시지 까지 모아서 요청 및 취득을 실시 -> 반복 -> `계속적인 메시지 취득`

* 작은 범위로 요청을 하는 경우

  > 하나의 메시지 취득 : 메시지마다 `Current Offset`을 업데이트

#### 컨슈머의 롤백

`Offset Commit`의 구조를 이용해 컨슈머 처리 실패, 고장시 -> `롤백 메시지 재취득`\
`Current Offset`과 `Offset Commit`의 위치로 재전송 지점 파악 -> 메시지 재취득\
고장의 감지, 복구에 대해서 카프카에서 제공되는 것은 아님 -> **Consumer API를 이용한 애플리케이션에서 대처 필요** -> 다행히 Spark Streaming 등 `카프카 연계 기능 제공하는 대부분 분산 처리 프레임워크`는 컨슈머의 고장이나 장애를 감지하여 `재실행하는 메커니즘`이 있어서 사용하면 됨

#### 메시지 전송 시 파티셔닝

프로듀서 -> 메시지 송신 -> 파티션으로 어떻게 보낼지 결정 -> 파티셔닝 기능 제공 메시지 내에 `Key,Value` 중 Key값에 따라 송신 1. Key -> 해시값사용 메시지 `key를 명시적`으로 지정 -> key에 따라 파티션 결정 -> `동일 Key는 동일 ID파티션에 송신` 2. Key -> 라운드 로빈 사용 key에 null 입력 -> 지정하지 않음 -> 라운드로빈으로 파티셔닝 처리

* 해시 이용할 때, Key가 부족할 경우, 편향이 발생 -> 리소스 부분적으로 사용할 수 없는 상태 발생  
* 파티셔닝 로직 구현 -> `DefaultPartitioner 클래스`를 사용하여 구현
* Producer API 제공 -> `Partitioner 인터페이스 구현` -> **Key, Value 값에 따라 송신 로직 커스텀**

  ```java
  package org.apache.kafka.clients.producer;
  ... 
  public interface Partitioner extends Configurable, Closeable {
  public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster);

  public void close();
  }
  ```

#### 브로커의 데이터 보존 기간

브로커에 보관된어 있는 토픽 데이터 보관, 어떻게 삭제?\
`삭제된 데이터는 재취득이 불가`

1. 오래된 메시지 삭제

   오래된것 부터 삭제, 2가지 설정 가능

   1. 데이터 취득 후 `경과시간` : 시간,분,밀리초 등 설정. 지정된 시간 지나면 삭제 (기본 1주)
   2. `데이터 크기` : 축적데이터 > 지정한 데이터크기 경우 삭제. 기본 -1(크기 제한 없음)
2. 압축

   최신 Key의 데이터를 남겨두고 `중복하는 Key의 오래된 메시지가 삭제됨`

   **cleanup.policy=delete or compact 로 설정**

#### 데이터의 견고성을 높이는 복제구조

* 실행 중 서버 고장 -> 수신메시지를 잃지 않기 위해, `복제(Replication)구조`를 갖춤
* **토픽 단위로 레플리카 수 지정**
* `Leader`, `Follower` 두개의 개념으로 구성
* Follower(Slave)는 Leader (Master)를 따름
* 프로듀서/컨슈머와 데이터 교환은 `Leader` 담당

#### 순서보증

* 단일 파티션의 경우 보증 되지만, Kafka는 기본적으로 다중 파티션을 써야 장점  
* **메시지 발생시점과 무관하게 컨슈머의 소비시점에 따라 순서가 뒤바뀌는 경우 발생**  
* 순서보증 필요 ? -> `해시 값에 의한 파티셔닝` 고려
* 해시에 의한 분할 -> 전체메시지 아닌, 프로듀서 및 장치당 ID 등 `카테고리별 순서제어를 목표할 때 유용`
* 완전정렬 -> 구현난이도도 높고, 컨슈머/프로듀서의 종속성은 방향성에 안맞음

#### 레플리카의 동기 상태

* ISR(min.insync.replica) 설정이 가능?
* 고장등으로 인한 일시적인 동기 지연을 허용. 전체 읽고 쓰기를 계속하는것이 가능

#### 복제 완료 최신 오프셋(High Wartermark)

* 복제사용시, 오프셋 관리 -> LEO, High Wartermark 
* `High Wartermark` : 복제가 완료된 오프셋
* 컨슈머 -> High Wartermark 까지 기록된 메시지를 취득할 수 있음
* ISR 및 High Wartermark
* `replica.lag.time.max.ms`에서 정한 시간보다 오랫동안 복제의 요청 및 복제가 이루어지지 않을 경우 복제 상태를 유지하지 않는 것으로 간주됨
* LEO 기록 -> 복제완료되지 않은 메시지 취득 -> `Leader Replica`를 갖는 브로커 복제 완료 안됨 -> 이 타이밍에 고장 발생 -> `그 사이에 취득한 메시지는 두번 다시 취득 할 수 없는 상태로 됨`

#### 프로듀서의 메시지 전달 보증 수준

* 프로듀서의 메시지 송신시 `Ack` 설정
* 프로듀서 -> 브로커 메시지 송신 / `브로커 -> 프로듀서 Ack 송신`의 타이밍 설정
* 0 : `Ack를 기다리지 않고` 다음 메시지 송신
* 1 : `Leader Replica`에 메시지 전달되면 `Ack 반환`
* all : `모든 ISR`의 수만큼 복제되면 `Ack반환` -> 성능이 좀 떨어짐

> 프로듀서 : 타임아웃 설정 등, Ack가 오지 않으면 `송신 실패로 감지` Ack 1 or all 설정 -> 반환타이밍이 의미하는 것은 복제에 '메시지가 전달'된 것으로 판단하는 수준, 타이밍을 의미한다. 이 타이밍에 메시지가 디스크에 flush되는 것이 아니라 메모리(OS버퍼)에 기록됨. 디스크 영속화는 다른속성에서 제어를 함

#### In-Sync Replica와 Ack = all, 쓰기 계속성의 관계

브로커 4대, 레플리카수 3, 브로커1대 고장->레플리카도1개 잃어버린 상황

1. min.insync.replica=3(레플리카 수와 동일), Ack=all 인 경우

   브로커 서버가 1대 고장난 경우, 프로듀서는 비정상 상태로 간주->잃어버린 레플리카가 `ISR로 복귀할 때 까지` 데이터를 쓸 수 없음
2. min.insync.replica=2, Ack=all 인 경우

   브로커 서버가 1대 고장, `Ack를 반환하고 처리를 계속함`. 처리 계속하는 점은 1번 보다 나은 반면, 나중에 추가된 파티션이 복제를 완료해 ISR로 승격될 떄까지 복제수가 2가 됨. 복구 전에 2대가 고장난 경우는 처리 중인 메시지를 손실할 위험 높음 

> `min.insync.replica`의 설정은 시스템의 상황에 따라 결정해서 메시지의 흐름을 제어 해야 함

#### 정리

1. 스케일아웃구성

   브로커 수를 증가, 클러스터링 -> 처리량 증가
2. 디스크 영속화

   데이터 저장(기본 1주) -> 재취득 가능
3. 연계할 수 있는 제품 존재

   프로듀서/컨슈머 API 제공 -> OSS 제품 다양하게 존재
4. 메시지 전달 보증

   Ack, Offset Commit 방식 도입 -> 메시지 송수신 성공/실패 관리 및 전략