[Kafka] 카프카 핵심 가이드 - 2 (Windows WSL 카프카 설치)

Chapter 2 카프카 설치하기

JAVA 설치

OS: windows 11

리눅스를 권장하고 있기 때문에 윈도우의 wsl을 사용해 ubuntu를 사용해 kafka를 설치합니다.

wsl에 우분투 설치하는 방법은 이 블로그(https://hy2on.tistory.com/148)를 참고하였습니다.

- powershell 관리자모드에서 wsl --install을 입력하여 설치 --> 제 경우엔 별도의 설정없이 Ubuntu 24가 설치됨

- 'wsl --distribution Ubuntu' 또는 Ubuntu 앱으로 사용가능

• root 계정으로 들어간 뒤 'apt update'

sudu su # root 계정 접속
apt-get autoremove --purge # java 충돌을 피하기 위해 기존것 삭제 (본인 환경에 따라 조정)
sudo apt-get autoremove --purge 
apt-get autoclean
sudo apt-get update && sudo apt-get upgrade # 업데이트
apt-get install openjdk-11-jdk # 책과 동일하게 jdk 11버전 설치
java -version # jdk 11.0.27로 설치됨

다소 책과 다른 점은 java 설치 위치가 다르다는 것 주의해야함. 책은 'usr/java/'이지만 내 경우 apt-get을 사용하였기 때문에 '/usr/lib/jvm'에 위치하고 있음

# ls -l /etc/alternatives/java
lrwxrwxrwx 1 root root 43 Apr 17 17:40 /etc/alternatives/java -> /usr/lib/jvm/java-11-openjdk-amd64/bin/java

주키퍼 설치

카프카의 클러스터의 메타데이터 컨슈머 클라이언트에 대한 정보를 저장하기 위해 아파치 주키퍼를 사용함

주키퍼는 설정 정보 관리, 이름 부여, 분산 동기화, 그룹 서비스를 제공하는 중앙화된 서비스임

주키퍼 설치는 이 블로그(https://velog.io/@rntjdwns1030/4.-Zookeeper-%EC%84%A4%EC%B9%98)를 참고했다.

wget으로 원하는 버전을 다운로드(책과 동일하게 3.5.9)하고, /usr/local 폴더에 다운받고 tar 압축을 푼다.

cd /usr/local/ # 경로 이동
sudo wget https://archive.apache.org/dist/zookeeper/zookeeper-3.5.9/apache-zookeeper-3.5.9-bin.tar.gz # 다운로드
sudo tar -zxf apache-zookeeper-3.5.9-bin.tar.gz # 현 위치에서 압축해제
sudo mv apache-zookeeper-3.5.9-bin /usr/local/zookeeper # 폴더 이름 변경
mkdir -p /var/lib/zookeeper # 주키퍼의 데이터 파일을 저장할 디렉토리, 트랜잭션 로그나 스냅샷 등
cat > /usr/local/zookeeper/conf/zoo.cfg << EOF # 설정 파일 작성
> tickTime=2000
> dataDir=/var/lib/zookeeper
> clientPort=2181
> EOF

export JAVA_HOME=/usr/lib/jvm/java-11-openjdk-amd64 # 환경변수 설정
echo $JAVA_HOME # 적용 확인

시작

/usr/local/zookeeper/bin/zkServer.sh start # 시작하기

telnet으로 포트 확인하려는데 우분투에 telnet이 설치가 안되어 있어서 별도로 설치

설치는 이 블로그(https://mr-relax.tistory.com/entry/%EC%9A%B0%EB%B6%84%ED%88%AC%EC%99%80-centOS%EC%97%90-%ED%85%94%EB%84%B7Telnet-%EC%84%A4%EC%B9%98%ED%95%98%EA%B8%B0)를 참고하였음

- telnet 설치

apt install telnet
apt install telnetd
apt install xinetd

vim /etc/xinetd.conf 

# 맨끝에 추가
service telnet 
{ 
     disable = no 
     flag = REUSE 
     socket_type = stream 
     protocol = tcp 
     wait = no 
     user = root 
     server = /usr/sbin/in.telnetd 
     log_on_failure += USERID 
}

vim /etc/services
# telnet    23/tcp 옆에 telnetd 추가

# 실행
systemctl start xinetd

telnet으로 주키퍼 실행 확인

telnet localhost 2181
>Trying 127.0.0.1...
>Connected to localhost.
>Escape character is '^]'. # 당황하지 않고 아래 srvr을 입력
srvr
# 정보가 리턴됨
Zookeeper version: 3.5.9-83df9301aa5c2a5d284a9940177808c01bc35cef, built on 01/06/2021 19:49 GMT
Latency min/avg/max: 0/0/0
Received: 1
Sent: 0
Connections: 1
Outstanding: 0
Zxid: 0x0
Mode: standalone
Node count: 5
Connection closed by foreign host.

* 참고로 주키퍼 사용은 점차 사라지는 추세이며, 대신 Kraft라는 카프카 자체 기능을 사용하는 것으로 바뀌고 있다.

 

카프카 설치

sudo wget https://archive.apache.org/dist/kafka/2.7.0/kafka_2.13-2.7.0.tgz # 책의 예제용이므로 낮은 버전 설치
tar -zxf kafka_2.13-2.7.0.tgz
mv kafka_2.13-2.7.0 kafka
mkdir /tmp/kafka-logs
export JAVA_HOME=/usr/lib/jvm/java-11-openjdk-amd64 # 환경변수 설정
echo $JAVA_HOME # 적용 확인
/usr/local/kafka/bin/kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties # 실행

테스트

/usr/local/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --replication-factor 1 --partitions 1 --topic test
# 이 명령어는 localhost:9092에서 실행 중인 Kafka 브로커에 연결하여, 
# 복제 계수 1과 파티션 1개를 가진 test라는 이름의 토픽을 만듭니다.

/usr/local/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic test
# 이 명령어는 localhost:9092에서 실행 중인 Kafka 브로커에 연결하여 test 토픽에 대한 상세 정보를 출력합니다.
#(예: 파티션 수, 복제 계수, 각 파티션의 리더(Leader), 복제본(Replicas), ISR(In-Sync Replicas) 등)
# 이를 통해 토픽이 제대로 생성되었는지, 어떤 상태인지 확인할 수 있습니다.

/usr/local# /usr/local/kafka/bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test
>Test Message 1
>Test Message 2
>^C # 멈추기


/usr/local# /usr/local/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
Test Message 1
Test Message 2
Test Message 1
Test Message 2
^C  # 안멈추면 계속 출력됨
Processed a total of 4 messages

---

Kafka Producer 및 Consumer 명령어 풀이

제공해주신 두 가지 Kafka 명령어는 각각 **메시지를 생산(produce)**하고 **소비(consume)**하는 과정을 보여줍니다. Kafka의 핵심 기능을 직접 체험해본 것이라고 할 수 있습니다.

1. Kafka 콘솔 Producer (메시지 생산자)

/usr/local/kafka/bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test

이 명령어는 Kafka 클러스터의 test 토픽으로 메시지를 보내는 콘솔 기반의 생산자(producer)를 시작합니다.

• /usr/local/kafka/bin/kafka-console-producer.sh: Kafka 설치 경로(/usr/local/kafka) 안에 있는 bin 디렉터리에서 kafka-console-producer.sh 스크립트를 실행하겠다는 의미입니다. 이 스크립트는 터미널에서 직접 메시지를 입력하여 Kafka 토픽으로 전송할 수 있게 해줍니다.
• --bootstrap-server localhost:9092: Kafka 클러스터에 연결하기 위한 부트스트랩 서버 정보를 지정합니다. Kafka 생산자가 메시지를 보낼 브로커를 찾을 때 사용합니다.
• --topic test: 메시지를 보낼 토픽의 이름을 test로 지정합니다.

실행 과정:

명령어를 실행하면 터미널 프롬프트가 >로 바뀝니다. 이때 입력하는 각 줄이 test 토픽으로 전송되는 하나의 메시지가 됩니다.

• >Test Message 1
• >Test Message 2

위와 같이 입력하면 "Test Message 1"과 "Test Message 2"라는 두 개의 메시지가 test 토픽으로 전송됩니다.

^C는 Ctrl+C를 눌러 생산자 프로세스를 종료했음을 의미합니다.

요약: 이 명령어는 localhost:9092에 있는 Kafka 브로커의 test 토픽으로 사용자가 직접 입력하는 텍스트 메시지를 실시간으로 전송하는 도구를 실행합니다.

2. Kafka 콘솔 Consumer (메시지 소비자)

/usr/local/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

이 명령어는 Kafka 클러스터의 test 토픽에서 메시지를 읽어오는 콘솔 기반의 소비자(consumer)를 시작합니다.

• /usr/local/kafka/bin/kafka-console-consumer.sh: Kafka 설치 경로(/usr/local/kafka) 안에 있는 bin 디렉터리에서 kafka-console-consumer.sh 스크립트를 실행하겠다는 의미입니다. 이 스크립트는 지정된 토픽의 메시지를 터미널에 출력해줍니다.
• --bootstrap-server localhost:9092: Kafka 클러스터에 연결하기 위한 부트스트랩 서버 정보를 지정합니다. Kafka 소비자가 메시지를 읽어올 브로커를 찾을 때 사용합니다.
• --topic test: 메시지를 읽어올 토픽의 이름을 test로 지정합니다.
• --from-beginning: 이 옵션이 중요합니다. 소비자가 토픽의 가장 처음부터 저장된 모든 메시지를 읽기 시작하도록 지시합니다. 이 옵션이 없으면 일반적으로 소비자는 자신이 시작된 시점 이후에 도착하는 새로운 메시지만을 읽게 됩니다.

실행 과정:

명령어를 실행하면 test 토픽에 저장된 메시지들이 순서대로 터미널에 출력됩니다.

• Test Message 1
• Test Message 2
• Test Message 1
• Test Message 2

여기서 "Test Message 1"과 "Test Message 2"가 두 번씩 나타난 것은, 이전에 생산자가 메시지를 전송하고 종료한 후, 다시 소비자를 시작하면서 --from-beginning 옵션 때문에 이미 저장된 메시지들을 다시 읽었기 때문입니다.

^CProcessed a total of 4 messages는 Ctrl+C를 눌러 소비자 프로세스를 종료했으며, 총 4개의 메시지를 처리했음을 나타냅니다.

요약: 이 명령어는 localhost:9092에 있는 Kafka 브로커의 test 토픽에서 가장 처음부터 모든 메시지를 읽어와 터미널에 출력하는 도구를 실행합니다.

---

Q. 부트스트랩 서버란?

: Kafka에서 부트스트랩 서버 (Bootstrap Server)는 Kafka 클라이언트(생산자 또는 소비자)가 Kafka 클러스터에 처음 연결할 때 사용하는 최초 진입점 역할을 하는 서버를 의미함 ==> 예: IP와 포트번호

브로커 주요 매개변수 설정하기

이 값들은 config/server.properties 파일에 설정할 수 있음

매개변수	설명	필수여부	기본값
broker.id	클러스터 내에서 각 브로커를 식별하는 고유한 정수 ID. 각 브로커는 반드시 고유한 ID를 가져야 하며, 이 ID는 클러스터 내부 통신 및 메타데이터 관리에서 중요한 역할	필수	없음
listeners	브로커가 클라이언트 연결을 수신할 네트워크 인터페이스와 포트를 정의. PLAINTEXT://호스트명:포트 또는 PLAINTEXT://IP주소:포트 형식으로 지정. PLAINTEXT://0.0.0.0:9092는 모든 IP 주소에서 9092 포트로 연결을 허용	필수	없음
advertised.listeners	클라이언트가 이 브로커에 연결하기 위해 사용하는 외부 주소(호스트명 또는 IP)와 포트를 정의. 일반적으로 listeners와 동일하게 설정하지만, 외부 접근이 필요할 때 다르게 설정.	선택	listeners와 동일
zookeeper.connect	ZooKeeper 앙상블에 연결할 주소를 지정. 호스트명1:포트1,호스트명2:포트2,... 형식으로 여러 ZooKeeper 서버를 콤마로 구분하여 나열할 수 있음. 카프카 클러스터의 메타데이터를 ZooKeeper가 관리하므로 이 설정은 필수적임.	필수	없음
log.dirs	카프카 메시지 로그 파일(토픽 파티션 데이터)이 저장될 로컬 파일 시스템 경로를 지정.	필수	/tmp/kafka-logs (테스트용)
num.recovery.threads.per.data.dir	브로커가 시작되거나 종료될 때 데이터 디렉토리당 로그 복구에 사용될 스레드 수. 디스크 I/O가 병목이 될 수 있는 환경에서 복구 성능을 조절	선택	1
auto.create.topics.enable	프로듀서가 존재하지 않는 토픽으로 메시지를 보내거나 컨슈머가 존재하지 않는 토픽을 구독할 때 자동으로 토픽을 생성할지 여부를 결정. 운영 환경에서는 실수로 토픽이 생성되는 것을 방지하기 위해 false로 설정하는 것이 일반적	선택	true
auto.leader.rebalance.enable	리더 파티션이 불균형하게 분포되었을 때 브로커가 자동으로 리더 파티션을 재분배할지 여부를 결정. 일반적으로 false로 유지하고 필요시 수동으로 재분배하는 것을 권장.	선택	false
delete.topic.enable	카프카 클러스터에서 kafka-topics.sh --delete 명령어를 사용하여 토픽을 실제로 삭제할 수 있도록 허용할지 여부를 결정. false로 설정하면 delete 명령이 작동하지 않음. 운영 환경에서는 데이터 손실을 막기 위해 false로 설정하는 경우가 많음	선택	true

---

 

브로커의 개수

카프카 클러스터의 적절한 크기를 결정하는 요소는 주로, 디스크 용량, 브로커당 레플리카 용량, CPU 용량,네트워크 요량을 고려한다.

1) 필요한 메시지를 저장하는 데 필요한 디스크 용량과 단일 브로커가 사용할 수 있는 저장소 용량을 고려한다. 만약, 클러스터가 10TB를 저장하고 있어야 하는데 하나의 브로커가 저장할 수 있는 용량이 2TB라면, 클러스터의 최소 크기는 브로커 5대가 된다(클러스터의 크기는 갯수를 의미하는 것인가?)

2) 레플리카는 하나의 파티션이 복제되는 서로 다른 브로커의 수를 의미한다. 즉, 위 예시와 같은 클러스터에서 복제 팩터를 2로 잡아 주면 최소한 10대의 브로커가 필요하다.

3) 클러스터가 처리 가능한 요청한 양은 CPU, 네트워크를 고려한다. 주로 네트워크의 영향을 많이 받는다. 네트워크 인터페이스의 전체 용량, 데이터를 읽거나 보존되는 동안 트래픽이 일정하지 않을 경우에도 클라이언트 트래픽을 받아낼 수 있는지를 체크해야한다. 

 

다수의 카프카 브로커가 하나의 클러스터를 이루게 하기 위해선 주키퍼 설정을 함께 해주어야한다.

1) zookeeper.connect 설정값. 클러스터가 메타데이터를 저장하는 주키퍼 앙상블과 경로 지정

2) 클러스터 안의 모든 브로커가 유일한 broker.id 설정값을 가져야함

 

가상 메모리

처리량이 중요한 애플리케이션은 스와핑을 막는 것이 최선이다. 메모리의 페이지가 디스크로 스와핑되는 과정에서 발생하는 비용이 발생하며 영향을 미치기 때문이다. 카프카는 시스템 페이지 캐시를 매우 많이 사용하기 때문에 가상 메모리 시스템이 디스크로 페이지로 스와핑할 경우 페이지 캐시를 할당할 메모리가 충분하지 않게 된다. 

 

더티 페이지(Dirty Page)란?

컴퓨터의 메모리 관리 시스템에서 더티 페이지는 메인 메모리(RAM)에 로드된 페이지(일반적으로 4KB 등의 고정된 크기 단위) 중, 원본 디스크 이미지와 내용이 달라진 페이지를 의미합니다. 즉, 디스크에서 읽어온 후 내용이 수정되었지만 아직 디스크에 다시 쓰여지지 않은 페이지입니다.

• 더티 페이지: RAM에 있지만, 디스크의 원본과 내용이 달라져 아직 디스크에 쓰여지지 않은 데이터 페이지. 카프카의 쓰기 버퍼(페이지 캐시)에 해당.
• spike 스파이크: 더티 페이지가 너무 많이 쌓여 한꺼번에 디스크로 플러시될 때 발생하는 급격한 I/O 부하 및 성능 저하.
• 버퍼링: 페이지 캐시를 활용하여 디스크 쓰기 작업을 일시적으로 메모리에 보관(버퍼링)함으로써 I/O 성능을 최적화하는 기법. 더티 페이지는 이 버퍼링의 결과물.

카프카와 같은 고처리량 시스템에서는 더티 페이지의 양을 적절히 관리하여 스와핑을 방지하고, 불필요한 I/O 스파이크를 최소화하는 것이 매우 중요합니다. 이를 위해 리눅스 커널의 vm.dirty_background_ratio, vm.dirty_ratio, vm.swappiness 등의 파라미터를 튜닝하기도 합니다.

 

가비지 컬렉터(Garbage Collector)란?

가비지 컬렉터는 프로그래머가 명시적으로 메모리를 해제하지 않아도, 더 이상 사용되지 않는 객체(쓰레기, Garbage)들을 자동으로 찾아내어 메모리에서 제거하고, 그 공간을 재활용하여 메모리를 효율적으로 관리하는 JVM의 핵심 구성 요소

 

1. 가비지 컬렉터의 동작 방식 (개념적으로):

1. Stop-the-World (STW): 대부분의 가비지 컬렉터는 GC 작업을 수행하는 동안 애플리케이션 스레드의 실행을 일시적으로 중단시킵니다. 이 중단 시간을 "Stop-the-World"라고 부르며, 이 시간이 길어질수록 애플리케이션의 응답성이 떨어집니다. 가비지 컬렉터의 목표 중 하나는 이 STW 시간을 최소화하는 것입니다.
2. 객체 추적 (Mark): 힙(Heap) 메모리에 있는 객체들 중, 현재 애플리케이션에서 사용되고 있는 객체(reachable objects)들을 식별하고 표시(mark)합니다. 이는 main 메소드나 스택에 있는 참조 변수(GC Root)에서 시작하여 참조되는 모든 객체를 따라갑니다.
3. 객체 제거 (Sweep): Mark 단계에서 표시되지 않은 객체들, 즉 더 이상 사용되지 않는 객체(unreachable objects, 가비지)들을 메모리에서 제거합니다.
4. 객체 압축 (Compact, Optional): 제거된 객체들로 인해 생긴 빈 공간(프래그먼테이션)을 없애기 위해 사용 중인 객체들을 메모리 한쪽으로 모으는(압축) 작업을 수행하기도 합니다. 이는 새로운 객체 할당 시 더 큰 연속된 공간을 제공하여 효율성을 높입니다.

2. G1GC(Garbage-First Garbage Collector)란?

• 등장 배경: 대규모 힙(수십 GB 이상) 환경에서 기존 GC들의 긴 STW 시간, 메모리 단편화 문제 해결을 위해 도입되었습니다. (Java 9부터 기본 GC)
• 주요 특징:
  • Region 기반 힙 분할: 힙을 작은 영역(Region)들로 나누고, 각 Region의 역할을 동적으로 할당합니다.
  • 가비지 우선(Garbage-First): GC 시 가장 많은 쓰레기(가비지)가 있는 Region부터 우선적으로 수집하여 효율을 높입니다.
  • 예측 가능한 STW 목표: 개발자가 설정한 최대 STW 시간(Pausetime Goal)을 지키기 위해 GC 작업을 조절합니다.
  • 동시 압축: GC 과정에서 메모리 압축을 동시에(Concurrent) 수행하여 단편화를 효과적으로 방지합니다.
• 동작 방식: 여러 단계(Initial Mark, Concurrent Marking, Remark, Cleanup, Copying/Evacuation)를 거치며, 일부는 STW를 발생시키고 일부는 애플리케이션과 동시에 진행됩니다.