[AWS] EC2를 이용한 KAFKA 클러스터 구축

이번 시간에는 AWS EC2를 이용하여 Kafka 클러스터를 구축해볼것 이다. 그전에 kafka 에 대해 알아보자!

- 분산 메세징 큐 시스템

- 대용량 처리에 특화되어 기존 메세징 시스템 보다 우수한 TPS를 보여줌

- 단순한 메시지 헤더를 지닌 TCP 기반의 프로토콜 사용 (rabitmq는 APMQ 프로토콜 사용)

개인 적인 의견으로 빠르게 구축하고 싶으면 rabbitmq 사용 (또한, 모니터링 툴도 함께 제공하고 있어 따로 구축하지 않아도 된다.), 빅데이터 처리 또는 빠른 처리를 위한것이면 kafka 를 사용하는 것이 좋다고 본다. 

처리 속도 비교

아래 용어에 대해 먼저 간단하게 알아보자!

개념	설명
topic	분류된 메세지의 집합
producer	메세지 등록하는 어플리케이션
consumer	메세지를 가져오는 어플리케이션
partition	토픽을 분산 처리 하기 위한 기술 (토픽을 파티션으로 나눠 처리)
replication	데이터 손실을 방지하기 위한 기술 (복제)
consumer group	메세지 처리하는 컨슈머 그룹 그룹내 각각의 컨슈머는 하나의 토픽의 다른 파티션의 내용만 처리할 수 있다.
zookeeper	클러스터 관리 (클러스터 최신 설정 정보 관리, 리더 선출, 토픽&파티션&밸런스 관리 등)
quorum	주키퍼에서 채택한 방식 의사결정을 하는데 최소한의 인원수 서버의 개수를 n개라고 했을 때 쿼럼의 수는 n/2 + 1개
offset	메세지 큐 위치 값
rebalance	컨슈머가 처리 불가시 해당 컨슈머가 담당하던 파티션을 다른 컨슈머가 처리할 수 있도록 하는 과정
ISR(In sync replica)	replication group 리더들은 팔로워들이 데이터를 주기적으로 확인하고 있는지 확인 하여 일정 주기(replica.lag.time.max.ms) 내에 요청이 오지 않는 다면 ISR 그룹에서 추방 한다. ISR 그룹에 속한 팔로워가 리더 자격을 갖는다.

용어를 익혔다면 아래 개념에 대해서도 익힌다!

• 브로커가 메세지를 보내주는것이 아닌 컨슈머가 데이터를 가져와 처리하는 방식
• 브로커 내에 처리한 메세지도 일정 기한 동안 보관한다. (log.retention.hours 설정 : 기본 7일)
• 파티션은 한번 추가되면 삭제를 할 수 없다.(삭제하고 싶다면 토픽을 제거하고 재 생성)
• 프로듀서가 같은 키로 보내면 같은 파티션으로 보내지며, 키를 지정 하지 않으면 라운드 로빈 동작으로 파티션이 지정 되어 전송한다.
• 파티션내 순서는 보장하며, offset이라는 단위로 고유 id가 증가한다.

자 이제 개념을 알았다면 구축해본다!

우선 아키텍쳐는 아래와 같으며 서버 3대로 클러스터 구성하며 서버 각각 zookeeper , broker 하나씩 띄우도록 한다

우선 AWS 에서 EC2 인스턴스를 생성 해보자!

1단계 : 생성하고자 하는 템플릿을 선택한다. 

2단계 : 인스턴스 유형도 cpu, mem 를 보고 원하는 스펙으로 선택한다.

3단계 : 인스턴스 개수는 3대를 한번에 만들기 위해 3으로 적으며 원하는 네트워크와 서브넷을 선택하고 퍼블릭 IP 사용예정으로 활성화를 선택한다.

3,

4단계 : 스토리지도 원하는 스펙으로 선택한다.

5단계 : 검색을 쉽게 하기위해 태그를 추가한다. (태그 key, value 로 검색이 가능하다.)

6단계 : 보안 그룹은 방화벽으로 아래와 같이 필요한 포트에 대해 방화벽을 허용한다. 아래와 같이 위치무관으로 설정하면 모두 허용하는 것으로 보안 이슈가 있어 추후 아래에 나오는 참고 자료와 같이 수정한다.

ssh를 제외하고 위치 무관이 아닌 kafka 서버 IP 대역을 넣는다. (예 : 1.1.1.0/24)

보안 그룹 생성한뒤 keypair 생성 까지 완료 하면 끝난다! (pem 파일은 서버 접속시 사용한다.)

<참고> 인스턴스 생성후 IP가 할당 되면 아래와 같이 보안 그룹을 수정하면 된다.

생성한뒤 정상적으로 인스턴스 3개가 떠있는것을 볼 수 있다.

만든 EC2 서버에 접속하는 방법은 어렵지 않다!

다운 받은 pem을 저장한 디렉토리로 가서 아래와 같이 접속하면 에러가 발생한다...

(base) hongyoolee@hateyou kafka % ssh -i kafka-test.pem ec2-user@X.X.X.X
@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
@         WARNING: UNPROTECTED PRIVATE KEY FILE!          @
@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
Permissions 0644 for 'kafka-test.pem' are too open.
It is required that your private key files are NOT accessible by others.
This private key will be ignored.
Load key "kafka-test.pem": bad permissions
ec2-user@X.X.X.X: Permission denied (publickey,gssapi-keyex,gssapi-with-mic).

원인을 찾아보니 pem 파일에 대한 권한이 없어서로 권한을 600으로 변경한뒤 다시 접속해본다

(base) hongyoolee@hateyou kafka % ls -all
total 4232
drwxr-xr-x   4 hongyoolee  staff      128  1  8 12:34 .
drwxr-xr-x  23 hongyoolee  staff      736  8  8 12:41 ..
-rw-r--r--@  1 hongyoolee  staff     1704  1  8 12:28 kafka-test.pem
-rwxr-xr-x@  1 hongyoolee  staff  1253339  5  2  2021 kafka.key

(base) hongyoolee@hateyou kafka % chmod 600 kafka-test.pem
(base) hongyoolee@hateyou kafka % ls -all
total 4232
drwxr-xr-x   4 hongyoolee  staff      128  1  8 12:34 .
drwxr-xr-x  23 hongyoolee  staff      736  8  8 12:41 ..
-rw-------@  1 hongyoolee  staff     1704  1  8 12:28 kafka-test.pem
-rwxr-xr-x@  1 hongyoolee  staff  1253339  5  2  2021 kafka.key

정상적으로 접속이 되는 것을 볼 수 있다!

(base) hongyoolee@hateyou kafka % ssh -i kafka-test.pem ec2-user@x.x.x.x

       __|  __|_  )
       _|  (     /   Amazon Linux 2 AMI
      ___|\___|___|

https://aws.amazon.com/amazon-linux-2/
6 package(s) needed for security, out of 16 available
Run "sudo yum update" to apply all updates.
[ec2-user@ip-x-x-x-x ~]$

생성한 인스턴스의 스펙이 맞는지 확인 해보면 맞게 생성되었다!

[ec2-user@ip-x-x-x-x ~]$ top
top - 03:45:37 up 15 min,  1 user,  load average: 0.00, 0.00, 0.00
Tasks:  93 total,   1 running,  50 sleeping,   0 stopped,   0 zombie
%Cpu(s):  0.0 us,  0.0 sy,  0.0 ni,100.0 id,  0.0 wa,  0.0 hi,  0.0 si,  0.0 st
KiB Mem :  2019796 total,  1461372 free,    82436 used,   475988 buff/cache
KiB Swap:        0 total,        0 free,        0 used.  1796420 avail Mem

[ec2-user@ip-x-x-x-x ~]$ grep -c processor /proc/cpuinfo
1

hostname 은 너무 어렵게 되어 있어서 서버 구분을 쉽게 하기위해 이름을 변경한다! (기호에 맞게 설정 하며 필수는 아니다.)

[ec2-user@ip-x-x-x-x ~]$ hostname
ip-x-x-x-x.ap-northeast-2.compute.internal
// 편의를 위해 변경 (kafka-01~03 까지 변경)
[ec2-user@ip-172-31-50-172 ~]$ sudo hostnamectl set-hostname kafka-01
[ec2-user@ip-172-31-50-172 ~]$ hostname
kafka-01

설정 및 관리를 쉽게 하기위해 생성한 인스턴스 host를 설정하며 ip는 private IP 로 설정 한다.

[ec2-user@ip-172-31-50-172 etc]$ sudo vi hosts
127.0.0.1   localhost localhost.localdomain localhost4 localhost4.localdomain4
::1         localhost6 localhost6.localdomain6
x.x.x.x kafka-01
x.x.x.x kafka-02
x.x.x.x kafka-03
:wq!

자 이제 kafka 가 동작하기 위해서 필요한 zookeeper 를 설치한다!

- zookeeper 버전 확인 : https://zookeeper.apache.org/releases.html

Apache ZooKeeper

위 사이트를 접속하여 설치 가능한 버전을 확인 하여 원하는 버전의 bin 파일 다운로드 주소를 복사한다. (저자의 경우 안정화된 버전인 3.6.3을 설치한다.)

source release 가 아닌 링크를 클릭하여 bin 파일을 다운 받는다. (source release로 다운 받아 진행시 오류 발생)

[ec2-user@x.x.x.x]$ wget https://dlcdn.apache.org/zookeeper/zookeeper-3.6.3/apache-zookeeper-3.6.3-bin.tar.gz
[ec2-user@x.x.x.x]$ tar xvf apache-zookeeper-3.6.3-bin.tar.gz
[ec2-user@x.x.x.x]$ cd apache-zookeeper-3.6.3-bin/

주키퍼 설정 파일을 아래와 같이 수정한다.

[ec2-user@x.x.x.x apache-zookeeper-3.7.0]$ cd conf
[ec2-user@x.x.x.x conf]$ ll
-rw-r--r-- 1 ec2-user ec2-user  535  3월 17  2021 configuration.xsl
-rw-r--r-- 1 ec2-user ec2-user 3435  3월 17  2021 log4j.properties
-rw-r--r-- 1 ec2-user ec2-user 1148  3월 17  2021 zoo_sample.cfg
[ec2-user@x.x.x.x conf]$ cp zoo_sample.cfg zoo.cfg
[ec2-user@x.x.x.x conf]$ vi zoo.cfg

설정 파일에 설정한 로그 파일 경로를 생성한다.

[ec2-user@ip-x.x.x.x conf]$ sudo mkdir /data/app_logs/zookeeper
[ec2-user@ip-x.x.x.x app_logs]$ sudo chmod -R 777 /data

설정중 server.1 은 dataDir 의 myid 파일에 데이터가 1인 파일이 있어야 하며 아래와 같이 myid 파일을 생성한다.

# kafka-01 서버에서 실행
echo 1 > /data/app_logs/zookeeper/myid
# kafka-02 서버에서 실행
echo 2 > /data/app_logs/zookeeper/myid
# kafka-03 서버에서 실행
echo 3 > /data/app_logs/zookeeper/myid

zookeeper 는 자바 기반 어플리케이션이므로 자바 설치 해야한다.

[ec2-user@ip-x.x.x.x ~]$ java -V
-bash: java: command not found

설치 가능한 라이브러리를 확인하여 java 1.8에 해당하는 openjdk 이름을 복사한다. 

$ yum list *openjdk*

설치를 진행하고 자바 버전을 확인하면 정상적으로 설치 된것을 볼 수 있다.

[ec2-user@ip-x.x.x.x ~]$ sudo yum -y install java-1.8.0-openjdk.x86_64
Complete!
[ec2-user@ip-x.x.x.x ~]$ java -version
openjdk version "1.8.0_312"
OpenJDK Runtime Environment (build 1.8.0_312-b07)
OpenJDK 64-Bit Server VM (build 25.312-b07, mixed mode)

쉽게 어플리 케이션을 관리하기 위해 alias 를 설정 하고 zookeeper 을 기동 및 중단 해본다.

[ec2-user@ip-x-x-x-x ~]$ vi ~/.bashrc
alias start_zookeeper='/home/ec2-user/apache-zookeeper-3.6.3-bin/bin/zkServer.sh start'
# 적용
[ec2-user@ip-x-x-x-x ~]$ source ~/.bashrc
[ec2-user@ip-x-x-x-x ~]$ alias
..
alias start_zookeeper='/home/ec2-user/apache-zookeeper-3.6.3-bin/bin/zkServer.sh start'
alias stop_zookeeper='/home/ec2-user/apache-zookeeper-3.6.3-bin/bin/zkServer.sh stop'
..
# 시작
[ec2-user@ip-x-x-x-x ~]$ start_zookeeper
ZooKeeper JMX enabled by default
Using config: /home/ec2-user/apache-zookeeper-3.6.3-bin/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED

# 주키퍼 시작시 dataDir 경로에 pid 파일이 생기고 주키퍼 중단시 삭제 된다.
[ec2-user@ip-x.x.x.x ~]$ cd /data/app_logs/zookeeper
[ec2-user@ip-x.x.x.x zookeeper]$ ls
-rw-rw-r-- 1 ec2-user ec2-user  2  1월  8 05:19 myid
drwxrwxr-x 2 ec2-user ec2-user 65  1월  8 05:19 version-2
-rw-rw-r-- 1 ec2-user ec2-user  5  1월  8 05:35 zookeeper_server.pid

# 종료
[ec2-user@ip-x-x-x-x ~]$ stop_zookeeper
ZooKeeper JMX enabled by default
Using config: /home/ec2-user/apache-zookeeper-3.6.3-bin/bin/../conf/zoo.cfg
Stopping zookeeper ... /home/ec2-user/apache-zookeeper-3.6.3-bin/bin/zkServer.sh: line 213: kill: (24901) - No such process
STOPPED

# 주키퍼 시작시 dataDir 경로에 pid 파일이 생기고 주키퍼 중단시 삭제 된다.
[ec2-user@ip-x.x.x.x ~]$ cd /data/app_logs/zookeeper
[ec2-user@ip-x.x.x.x zookeeper]$ ls
-rw-rw-r-- 1 ec2-user ec2-user  2  1월  8 05:19 myid
drwxrwxr-x 2 ec2-user ec2-user 65  1월  8 05:19 version-2

// 정상 실행 확인 방법

// 정상적으로 기동된 경우 프로세스가 떠있다.
$ ps -ef | grep zookeeper
ec2-user  8671     1 22 08:55 pts/0    00:00:01 /usr/lib/jvm/java-1.8.0-openjdk-1.8.0.312.b07-1.amzn2.0.2.x86_64/jre/bin/java -Dzookeeper.log.dir=/home/ec2-user/apache-zookeeper-3.6.3-bin/bin/../logs -Dzookeeper.log.file=zookeeper....
// 클라이언트 접속 (오류없이 실행되며 조회 가능한 주키퍼 쉘이 나온다.)
$ /home/ec2-user/apache-zookeeper-3.6.3-bin/bin/zkCli.sh
...
2022-01-08 08:57:19,769 [myid:localhost:2181] - INFO  [main-SendThread(localhost:2181):ClientCnxn$SendThread@1448] - Session establishment complete on server localhost/127.0.0.1:2181, session id = 0x100012a10de0000, negotiated timeout = 30000

WATCHER::

WatchedEvent state:SyncConnected type:None path:null
[zk: localhost:2181(CONNECTED) 0]
[zk: localhost:2181(CONNECTED) 0] quit

자 이제 kafka를 설치해본다!

- 다운로드 주소 : https://kafka.apache.org/downloads

 

위에 사이트에서 원하는 버전의 다운로드 주소를 복사하여 다운로드 하여 압축을 푼다

$ wget https://dlcdn.apache.org/kafka/3.0.0/kafka_2.12-3.0.0.tgz
$ tar xvf kafka_2.12-3.0.0.tgz

// kafka_2.12-3.0.0/config/server.properties

# kafka-01
broker.id=1
listeners=PLAINTEXT://:9092
advertised.listeners=PLAINTEXT://kafka-01:9092
zookeeper.connect=kafka-01:2181,kafka-02:2181,kafka-03:2181

# kafka-02
broker.id=2
listeners=PLAINTEXT://:9092
advertised.listeners=PLAINTEXT://kafka-02:9092
zookeeper.connect=kafka-01:2181,kafka-02:2181,kafka-03:2181

# kafka-03
broker.id=2
listeners=PLAINTEXT://:9092
advertised.listeners=PLAINTEXT://kafka-03:9092
zookeeper.connect=kafka-01:2181,kafka-02:2181,kafka-03:2181

위와 같이 서버별로 설정 하였다면 아래와 같이 kafka 관련 alias 를 등록한다.

// 백그라운드 실행은 -daemon을 붙여주세요
$ vi ~/.bashrc
alias start_kafka='/home/ec2-user/kafka_2.12-3.0.0/bin/kafka-server-start.sh -daemon /home/ec2-user/kafka_2.12-3.0.0/config/server.properties'
alias stop_kafka='/home/ec2-user/kafka_2.12-3.0.0/bin/kafka-server-stop.sh -daemon /home/ec2-user/kafka_2.12-3.0.0/config/server.properties'
:wq!
$ source ~/.bashrc
$ alias
alias start_kafka='/home/ec2-user/kafka_2.12-3.0.0/bin/kafka-server-start.sh -daemon /home/ec2-user/kafka_2.12-3.0.0/config/server.properties'
alias stop_kafka='/home/ec2-user/kafka_2.12-3.0.0/bin/kafka-server-stop.sh -daemon /home/ec2-user/kafka_2.12-3.0.0/config/server.properties'

// 시작

$ start_kafka

// 프로세스 확인 

$ ps -ef | grep kafka
ec2-user 18745     1 10 09:12 pts/0    00:00:05 /usr/lib/jvm/java-1.8.0-openjdk-1.8.0.312.b07-1.amzn2.0.2.x86_64/jre/bin/java -Xmx1G -Xms1G -server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=3....

 

자 이제 정상적으로 뜨는 부분 확인 하였으면 토픽을 생성해본다!

// 브로커 서버3대 중 한대에서 실행
$ /home/ec2-user/kafka_2.12-3.0.0/bin/kafka-topics.sh --create \
--replication-factor 3 \
--partitions 2 \
--topic topic-A \
--bootstrap-server kafka-01:9092,kafka-02:9092,kafka-03:9092

// 확인 (3대 모두 리스트 확인시 생성한 topic-A가 나온다.)
$ /home/ec2-user/kafka_2.12-3.0.0/bin/kafka-topics.sh --list --bootstrap-server kafka-01:9092,kafka-02:9092,kafka-03:9092
topic-A

// topic-B도 생성
$ /home/ec2-user/kafka_2.12-3.0.0/bin/kafka-topics.sh --create \
--replication-factor 3 \
--partitions 1 \
--topic topic-B \
--bootstrap-server kafka-01:9092,kafka-02:9092,kafka-03:9092

// 확인 (3대 모두 리스트 확인시 생성한 topic-A가 나온다.)
$ /home/ec2-user/kafka_2.12-3.0.0/bin/kafka-topics.sh --list --bootstrap-server kafka-01:9092,kafka-02:9092,kafka-03:9092
topic-A
topic-B

아래는 저자가 겪었던 오류 상황과 해결 방안을 공유 한다.. 여러분들은 빠르게 해결 하기를..!

# 오류

[ec2-user@ip-x.x.x.x ~]$ /home/ec2-user/apache-zookeeper-x.x.x-.../zkServer.sh start
ZooKeeper JMX enabled by default
Using config: /home/ec2-user/apache-zookeeper-x.x.x-.../conf/zoo.cfg
Starting zookeeper ... FAILED TO START

 

1. zookeeper 설치시 bin 이 아닌 source release 로 설치 했을시

-  bin 으로 끝나는 설치파일을 다운해야 한다.

2. myid 파일을 dataDir 설정 경로에 만들지 않았을시

- 서버별로 다르게 지정한다.

 

자 이제 producer와 consumer 를 구성하여 생성한 토픽에 메세지를 넣고 빼본다!

우선 local 에 producer 와 consumer 를 구성할 것이기에 kafka host 를 설정 하고 방화벽도 내 로컬 PC IP 로 9092 포트 접속에 대해 열어둔다. (위에 설정 그대로 했다면 추가 설정하지 않아도 된다.)

# local PC host 설정

// 퍼블릭 IP 로 설정 (aws에는 보안그룹 내 IP로 9092 열어놔야한다.)
x.x.x.x kafka-01
x.x.x.x kafka-02
x.x.x.x kafka-03

# kafka producer / consumer 생성 (아래 소스코드 참고)

1. producer 생성
2. consumer 생성
3. consumer group 생성 (group 명 코드와 같이 지정시 자동으로 컨슈머 그룹이 생성 된다.)

:: https://github.com/works-code/kafka

GitHub - works-code/kafka: kafka 컨슈머, 브로커 스타터 코드

# 테스트 케이스

1. 토픽, 파티션, 컨슈머 그룹별 처리 방식 확인
2. 장애 상황 테스트
   1. 브로커 팔로워가 죽었을시
   2. 브로커 리더가 죽었을시
   3. 컨슈머 그룹중 한명이 죽었을시
   4. 주키퍼 x대가 죽었을시 
3. at least once (중복 발생 하는지) / at exactly once (정확히 한번 : 중복 발생 안하는지)

테스트 케이스	테스트 순서	명령어	결과
토픽, 파티션, 컨슈머 그룹별 처리 방식 확인	topic-A , B 생성	$ /home/ec2-user/kafka_2.12-3.0.0/bin/kafka-topics.sh --create \ --replication-factor 3 \ --partitions 2 \ --topic topic-A \ --bootstrap-server kafka-01:9092,kafka-02:9092,kafka-03:9092	$ /home/ec2-user/kafka_2.12-3.0.0/bin/kafka-topics.sh --list --bootstrap-server kafka-01:9092,kafka-02:9092,kafka-03:9092 topic-A topic-B > A, B 가 정상적으로 생성됨
		$ /home/ec2-user/kafka_2.12-3.0.0/bin/kafka-topics.sh --create \ --replication-factor 3 \ --partitions 1 \ --topic topic-B \ --bootstrap-server kafka-01:9092,kafka-02:9092,kafka-03:9092
	procedure 1개 프로세스 consumer 2개 프로세스 - topic-A (group-01)/topic-B(group-01) - topic-A (group-01)/topic-B(group-01)		topic-A : key 값에 따라 컨슈머 2개에서 처리 되며 key 값이 동일하면 한개의 프로세스에서 처리 되므로 순서가 보장 된다. (key 값을 없이 보내면 라운드로빈 방식으로 처리) topic-B : 컨슈머 1대 에서만 처리됨 > 파티션 한개당 컨슈머 그룹내 1개의 컨슈머와 매핑된다.
	procedure 1개 프로세스 consumer 2개 프로세스 - topic-A (group-01)/topic-B(group-01) - topic-A (group-01)/topic-B(group-02)		topic-A : 위와 동일 topic-B : 컨슈머 그룹이 달라 컨슈머 2대에서 처리되나 병렬처리가 아닌 중복 처리 (1개의 메세지당 group-01,02 모두에서 처리)
장애 상황 테스트	브로커 팔로워가 죽었을시	// 팔로워 죽이기 전 $ /home/ec2-user/kafka_2.12-3.0.0/bin/kafka-topics.sh --describe --bootstrap-server kafka-01:9092,kafka-02:9092,kafka-03:9092 Topic: topic-B Partition: 0 Leader: 2 Replicas: 2,3,1 Isr: 2,3,1 Topic: topic-A Partition: 0 Leader: 1 Replicas: 1,2,3 Isr: 1,2,3 Topic: topic-A Partition: 1 Leader: 2 Replicas: 2,3,1 Isr: 2,3,1 // 팔로워 죽인후 $ /home/ec2-user/kafka_2.12-3.0.0/bin/kafka-topics.sh --describe --bootstrap-server kafka-01:9092,kafka-02:9092,kafka-03:9092 Topic: topic-B Partition: 0 Leader: 2 Replicas: 2,3,1 Isr: 2,1 Topic: topic-A Partition: 0 Leader: 1 Replicas: 1,2,3 Isr: 1,2 Topic: topic-A Partition: 1 Leader: 2 Replicas: 2,3,1 Isr: 2,1 // 팔로워 기동 Topic: topic-B Partition: 0 Leader: 2 Replicas: 2,3,1 Isr: 2,1,3 Topic: topic-A Partition: 0 Leader: 1 Replicas: 1,2,3 Isr: 1,2,3 Topic: topic-A Partition: 1 Leader: 2 Replicas: 2,3,1 Isr: 2,1,3	서비스 이상 없음
	브로커 리더가 죽었을시		서비스 이상 없음 > 데이터 IN OUT이 리더를 통해 이뤄지니 다운타임이 있을 것이라 생각 했으나 서비스 이상이 없어 의문 (데이터 양을 늘려서 테스트 예정)
	컨슈머 그룹중 한명이 죽었을시		1s 동안 리벨런싱 후(해당 시간 동안에는 처리 중단) 다른 컨슈머가 처리 하다가 기존 컨슈머가 돌아왔을시 다시 리벨런싱 되어 기존 컨슈머가 처리한다.
	주키퍼 x대가 죽었을시		테스트 필요

 

# 옵션

// zookeeper

옵션명	설정	설명
tickTime	2000(ms)	주키퍼 기본 시간 단위
initLimit	10	- 팔로워가 리더와 초기에 연결하는 시간에 대한 타임아웃 10 ticks 로 'tickTime(2초) * initLimit(10) = 20초
syncLimit	5	- 팔로워가 리더와 동기화 하는데에 대한 타임아웃 5 ticks 로 'tickTime(2초) * initLimit(5) = 10초
dataDir	/data/app_logs/zookeeper	주키퍼 로그, 스냅샷 저장 경로
clientPort	2181	주키퍼 클라이언트가 서버로 연결하기 위해 사용하는 포트
server.1	server.1=kafka-01:2888:3888 server.2=kafka-02:2888:3888 server.3=kafka-03:2888:3888	주키퍼 클러스터 구성 정보 2888: zookeeper sync port 3888: zookeeper cluster port

// kafka

옵션명	설정	설명
broker.id	1	브로커 식별 아이디
listeners	PLAINTEXT://:9092	카브카 브로커 사용 포트
advertised.listeners	PLAINTEXT://kafka-01:9092
num.network.threads	3	네트워크 요청 처리하는데 사용하는 스레드수
num.io.threads	8	입출력에 사용하는 스레드수
socket.send.buffer.bytes	102400	소켓서버가 사용하는 송수신 버퍼 사이즈
socket.receive.buffer.bytes	102400
socket.request.max.bytes	104857600	서버가 받을수 있는 최대 요청 사이즈 (약 100MB)
log.dirs	/tmp/kafka-logs	브로커가 프로듀서로 받는 메세지 저장 위치 쉼표로 다양한 경로에 저장 가능하다. 항상 고르게 분산되지 않는다.
num.partitions	1	토픽당 파티션수
num.recovery.threads.per.data.dir	1	로그 디렉터리 기준으로 복구할때 사용되는 쓰레드수 log.dirs 가 3개 라면 '3*1=' 3개의 쓰레드가 사용된다. (장애에 빠르게 대응하고자 한다면 높은값 설정을 고려)
offsets.topic.replication.factor	1	consumer offset, transaction state 에 대한 replica 수 운영시 3 copy 추천
transaction.state.log.replication.factor	1
transaction.state.log.min.isr	1
log.retention.hours	168	브로커가 프로듀서로 받는 메세지 파일 보관 기간 168시간(7일)
log.segment.bytes	1073741824	브로커가 프로듀서로 받는 메세지 보관 크기(1GB)로 해당 크기만큼 차면 새 파일 생성
log.retention.check.interval.ms	300000	브로커가 프로듀서로 받는 메세지 삭제 여부 체크 주기 300000ms = 5분
zookeeper.connect	kafka-01:2181,kafka-02:2181,kafka-03:2181	카프카 메타정보 저장할 주키퍼 클러스터 연결 정보
zookeeper.connection.timeout.ms	18000	주키퍼 접속 시도 제한 시간 (18초)
group.initial.rebalance.delay.ms	0	코디네이터가 파티션 리밸런싱을 처음 수행할 때 더 많은 컨슈머들이 새로운 컨슈머 그룹에 조인할 때까지 대기하는 시간 운영 환경에서는 3초의 기본값 사용 추천