[kafka] 카프카 클러스터(Cluster) 구축

# 설치 환경

AWS EC2

1. 3개의 인스턴스

2. t2.micro

3. Amazon Linux

 

Kafka

1. Kafka_2.13-3.7.1 버전

2. Zookeeper가 아닌 KRaft 모드로 구축

 

---

 

# 다운로드

1. 각각의 인스턴스에서 모두 진행

# java 찾기
yum list java*

# java 설치
sudo yum install -y java-17-amazon-corretto.x86_64

# 설치 확인
java --version
javac --version

# javac 가 없다 나올 시
sudo yum install java-17-amazon-corretto-devel.x86_64

# java 환경 설정
export JAVA_HOME=/usr/lib/jvm/java-17-amazon-corretto.x86_64
export PATH=$JAVA_HOME/bin:$PATH

# 아파치 카프카 다운로드
wget https://downloads.apache.org/kafka/3.7.1/kafka_2.13-3.7.1.tgz

# 압축해제
tar zxv kafka_2.13-3.7.1.tgz

# 디렉토리 이동
mv kafka_2.13.3.7.1 /opt/

 

---

# server.properties 설정

1. 각각의 인스턴스에서 모두 진행

2. 인스턴스에 맞는 프라이빗 IPv4 주소를 사용하여 설정

 

# 해당 폴더로 이동
cd /opt/kafka_2.13-3.7.1/config/kraft

# vi 를 사용하여 수정
vi server.properties

 

 

kafka01

process.roles=broker,controller
node.id=1

# 3개의 인스턴스 프라이빗 주소를 위치에 맞게 입력 (아래는 예시)
controller.quorum.voters=1@172.0.0.1:9093,2@172.0.0.2:9093,3@172.0.0.3:9093

# Broker의 Listen Address
listeners=PLAINTEXT://172.0.0.1:9092,CONTROLLER://172.0.0.1:9093

inter.broker.listener.name=PLAINTEXT

# Producer, consumer를 위한 Listen Address
advertised.listeners=PLAINTEXT://172.0.0.1:9092

controller.listener.names=CONTROLLER
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL

# 네트워크 요청을 처리하는 Thread의 개수
num.network.threads=3

# 디스크 I/O를 처리 하는 Thread의 개수
num.io.threads=8

# send buffer
socket.send.buffer.bytes=102400

# receive buffer
socket.receive.buffer.bytes=102400

# request buffer
socket.request.max.bytes=104857600

# 로그 세그먼트를 저장한 경로.
log.dirs=/opt/kafka_2.13-3.7.1/logs/kraft-log

# 토픽의 기본 파티션 개수
num.partitions=3

num.recovery.threads.per.data.dir=1

# 토픽의 파티션 replication 개수
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=3

# 로그 세그먼트 삭제를 위한 로그 보관기간. 168시간 이후의 로그를 삭제
log.retention.hours=168

# 로그 세그먼트의 최대 사이즈. 이 사이즈에 도달하면 새로운 로그 세그먼트가 생성됨
log.segment.bytes=1073741824

# 로그 retention을 위한 체크 간격
log.retention.check.interval.ms=300000

 

kafka02

process.roles=broker,controller
node.id=2

# 3개의 인스턴스 프라이빗 주소를 위치에 맞게 입력 (아래는 예시)
controller.quorum.voters=1@172.0.0.1:9093,2@172.0.0.2:9093,3@172.0.0.3:9093

# Broker의 Listen Address
listeners=PLAINTEXT://172.0.0.2:9092,CONTROLLER://172.0.0.2:9093

inter.broker.listener.name=PLAINTEXT

# Producer, consumer를 위한 Listen Address
advertised.listeners=PLAINTEXT://172.0.0.2:9092

controller.listener.names=CONTROLLER
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL

# 네트워크 요청을 처리하는 Thread의 개수
num.network.threads=3

# 디스크 I/O를 처리 하는 Thread의 개수
num.io.threads=8

# send buffer
socket.send.buffer.bytes=102400

# receive buffer
socket.receive.buffer.bytes=102400

# request buffer
socket.request.max.bytes=104857600

# 로그 세그먼트를 저장한 경로.
log.dirs=/opt/kafka_2.13-3.7.1/logs/kraft-log

# 토픽의 기본 파티션 개수
num.partitions=3

num.recovery.threads.per.data.dir=1

# 토픽의 파티션 replication 개수
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=3

# 로그 세그먼트 삭제를 위한 로그 보관기간. 168시간 이후의 로그를 삭제
log.retention.hours=168

# 로그 세그먼트의 최대 사이즈. 이 사이즈에 도달하면 새로운 로그 세그먼트가 생성됨
log.segment.bytes=1073741824

# 로그 retention을 위한 체크 간격
log.retention.check.interval.ms=300000

 

kafka03

process.roles=broker,controller
node.id=3

# 3개의 인스턴스 프라이빗 주소를 위치에 맞게 입력 (아래는 예시)
controller.quorum.voters=1@172.0.0.1:9093,2@172.0.0.2:9093,3@172.0.0.3:9093

# Broker의 Listen Address
listeners=PLAINTEXT://172.0.0.3:9092,CONTROLLER://172.0.0.3:9093

inter.broker.listener.name=PLAINTEXT

# Producer, consumer를 위한 Listen Address
advertised.listeners=PLAINTEXT://172.0.0.3:9092

controller.listener.names=CONTROLLER
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL

# 네트워크 요청을 처리하는 Thread의 개수
num.network.threads=3

# 디스크 I/O를 처리 하는 Thread의 개수
num.io.threads=8

# send buffer
socket.send.buffer.bytes=102400

# receive buffer
socket.receive.buffer.bytes=102400

# request buffer
socket.request.max.bytes=104857600

# 로그 세그먼트를 저장한 경로.
log.dirs=/opt/kafka_2.13-3.7.1/logs/kraft-log

# 토픽의 기본 파티션 개수
num.partitions=3

num.recovery.threads.per.data.dir=1

# 토픽의 파티션 replication 개수
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=3

# 로그 세그먼트 삭제를 위한 로그 보관기간. 168시간 이후의 로그를 삭제
log.retention.hours=168

# 로그 세그먼트의 최대 사이즈. 이 사이즈에 도달하면 새로운 로그 세그먼트가 생성됨
log.segment.bytes=1073741824

# 로그 retention을 위한 체크 간격
log.retention.check.interval.ms=300000

 

---

# 로그 세그먼트 디렉토리 생성

1. 각각의 인스턴스에서 모두 진행

2. 토픽의 파티션 스트림 데이터가 저장된다.

 

mkdir -p /opt/kafka_2.13-3.7.1/logs/kraft-log

 

---

# 로그 세그먼트 스토리지 포맷

1. 각각의 인스턴스에서 모두 진행

2. random-uuid 를 매번 생성하지 않고 생성해서 나온 uuid를 다른 인스턴스에 그대로 사용한다.

# 해당 경로로 이동
cd /usr/opt/kafka_2.13-3.7.1/bin

# random uuid 생성 (주의! : 한 번해서 나온 값을 인스턴스끼리 공유함) 
./kafka-storage.sh random-uuid

# 스토리지 포맷
./kafka-storage.sh format -t {생성해서 나온 uuid} -c /opt/kafka_2.13-3.7.1/config/kraft/server.properties

 

 

---

# Kafka broker heap memory 축소

1. 각각의 인스턴스에서 모두 진행

2. t2.micro만 해당

# 환경 설정
export KAFKA_HEAP_OPTS="-Xmx400m -Xms400m"
echo $KAFKA_HEAP_OPTS

# 그 후에 인스턴스가 종료되고 나서도 쓸 수 있도록 .bashrc에 넣어준다.
vi ~/.bashrc

 

---

# 인바운드 규칙 설정

1. 9092, 9093 포트를 연다.

2. 프라이빗 IPv4 주소를 사용하고 있기 때문에 인바운드 규칙에 자신이 속한 보안 그룹을 추가

 

---

# 카프카 구동

1. 각각의 인스턴스에서 모두 진행

/opt/kafka_2.13-3.7.1/bin/kafka-server-start.sh -daemon /opt/kafka_2.13-3.7.1/config/kraft/server.properties

 

---

# 상태 확인

# cluster 상태 확인
/opt/kafka_2.13-3.7.1/bin/kafka-metadata-quorum.sh --bootstrap-server 172.31.7.189:9092,172.31.8.251:9092,172.31.13.117:9092 describe --status

# replication 상태 확인
/opt/kafka_2.13-3.7.1/bin/kafka-metadata-quorum.sh --bootstrap-server 172.31.7.189:9092,172.31.8.251:9092,172.31.13.117:9092 describe --replication

 

---

참고자료

[Kafka] 구축 해보기[1] - Cluster 설치

[Kafka] 구축 해보기[1] - Cluster 설치

[Kafka] Intro. + AWS에 Kafka 브로커 생성하기

[Kafka] Intro. + AWS에 Kafka 브로커 생성하기