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

Kafka 기본 개념

---

참조 : https://zeroco.tistory.com/105

[Kafka] 카프카 기본 개념 구조 이해하기

 

 

Node 정보

---

 - CentOS Stream release 8 x64

Hostname	IP
test-kafka01	172.16.10.151
test-kafka02	172.16.10.152
test-kafka03	172.16.10.153

 

 

 

설치 개요

---

• 총 3대의 Node를 이용하여 kafka cluster를 구성할 예정이다.
• kafka는 3.7.0 버전을 사용한다.
• zookeeper가 아닌 KRaft 모드로 구축 한다.
  • KRaft는 Kafka 2.8.0 부터 추가되었으며 Kafka 3.3.0 버전부터 정식 릴리즈 되었다.
• UI 관리 도구로 "UI for Apache Kafka"를 사용한다.
• 모니터링은 "Promethous + Grafana"를 사용하며 "kafka-exporter"로 수집한다.

 

 

 

다운로드

---

• 모든 노드에서 진행한다.

## 다운로드
wget https://downloads.apache.org/kafka/3.7.0/kafka_2.13-3.7.0.tgz

## 압축해제
tar zxv fkafka_2.13-3.7.0.tgz

## 디렉토리 이동
mv kafka_2.13-3.7.0 /usr/local/kafka

 

 

 

 

server.properties 설정

---

• 모든 노드에서 진행한다.
• IP는 각 노드에 맞게 수정한다.

경로 : /usr/local/kafka/config/kraft/server.properties

 

test-kafka01

process.roles=broker,controller
node.id=1
controller.quorum.voters=1@172.16.10.151:9093,2@172.16.10.152:9093,3@172.16.10.153:9093

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

inter.broker.listener.name=PLAINTEXT

# Producer, consumer를 위한 Listen Address
advertised.listeners=PLAINTEXT://172.16.10.151: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=/usr/local/kafka/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

 

 

test-kafka02

process.roles=broker,controller
node.id=2
controller.quorum.voters=1@172.16.10.151:9093,2@172.16.10.152:9093,3@172.16.10.153:9093
listeners=PLAINTEXT://172.16.10.152:9092,CONTROLLER://172.16.10.152:9093
inter.broker.listener.name=PLAINTEXT
advertised.listeners=PLAINTEXT://172.16.10.152:9092
controller.listener.names=CONTROLLER
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/usr/local/kafka/logs/kraft-log
num.partitions=3
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=3
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000

 

 

test-kafka03

process.roles=broker,controller
node.id=3
controller.quorum.voters=1@172.16.10.151:9093,2@172.16.10.152:9093,3@172.16.10.153:9093
listeners=PLAINTEXT://172.16.10.153:9092,CONTROLLER://172.16.10.153:9093
inter.broker.listener.name=PLAINTEXT
advertised.listeners=PLAINTEXT://172.16.10.153:9092
controller.listener.names=CONTROLLER
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/usr/local/kafka/logs/kraft-log
num.partitions=3
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=3
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000

 

 

 

 

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

---

• 모든 노드에서 진행한다.
• 토픽의 파티션 스트림 데이터가 저장된다.

mkdir -p /usr/local/kafka/logs/kraft-log

 

 

 

 

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

---

• 모든 노드에서 진행한다.
• 단, random-uuid 생성한 임의의 한곳의 노드에서 진행후 값을 얻는다.

경로 : //usr/local/kafka/bin

# random uuid 생성
./kafka-storage.sh random-uuid

# 스토리지 포맷
./kafka-storage.sh format -t BJmcZfo5R5Cywkpwj98vmQ -c /usr/local/kafka/config/kraft/server.properties

 

 

 

 

kafka 구동

---

• 모든 노드에서 진행한다.

직접 daemon으로 구동.

/usr/local/kafka/bin/kafka-server-start.sh -daemon /usr/local/kafka/config/kraft/server.properties

 

 

혹은 systemd 등록을 통한 구동

경로 : /usr/lib/systemd/system/kafka.service

[Unit]
Description=kafka
After=syslog.target
After=network.target

[Service]
Type=simple
Restart=on-failure
ExecStart=/usr/local/kafka/bin/kafka-server-start.sh /usr/local/kafka/config/kraft/server.properties
ExecStop=/usr/local/kafka/bin/kafka-server-stop.sh
RestartSec=10s

[Install]
WantedBy=multi-user.target

 

systemctl daemon-reload
systemctl start kafka

 

 

 

 

상태확인

---

 

cluster 상태확인

/usr/local/kafka/bin/kafka-metadata-quorum.sh --bootstrap-server 172.16.10.151:9092,172.16.10.152:9092,172.16.10.153:9092 describe --status

ClusterId:              BJmcZfo5R5Cywkpwj98vmQ
LeaderId:               2
LeaderEpoch:            78
HighWatermark:          511479
MaxFollowerLag:         0
MaxFollowerLagTimeMs:   312
CurrentVoters:          [1,2,3]
CurrentObservers:       []

 

 

replication  상태 확인

/usr/local/kafka/bin/kafka-metadata-quorum.sh --bootstrap-server 172.16.10.151:9092,172.16.10.152:9092,172.16.10.153:9092 describe --replication

NodeId	LogEndOffset	Lag	LastFetchTimestamp	LastCaughtUpTimestamp	Status  	
2     	511672      	0  	1716446895864     	1716446895864        	Leader  	
1     	511672      	0  	1716446895854     	1716446895854        	Follower	
3     	511672      	0  	1716446895854     	1716446895854        	Follower