Kafka 실습

Google Cloud 설정

• Compute Engine API 사용

• 인스턴스 만들기

  • 이름, 리전 , 영역 선택
  • cpu 8개, 코어 4개 메모리 32kb
  • 부팅디스크 - centos stream9 , 20GB
  • 액세스 범위 : 모든 Cloud API 에 대한 전체 허용
  • 방화벽 : HTTP(80), HTTPS(443) 허용

• 연결

  • 브라우저에서 연결 가능
    • 세션 타임 아웃이 존재하는 불편함
  • 맞춤 포트의 브라우저 창
  • 다른 전용 ssh 클라이언트

key

1) key 생성

• public, private key 생성
• 이름 지정 가능

ssh-keygen -t rsa -C "구글계정명"

2) key 등록

• .pub 파일 복사(모든 내용)
• compute engine -> 설정 -> 메타데이터 -> ssh key -> 키 추가
• 붙여넣기

ssh 접속

1) 커멘드 창
2) ssh 명령어 입력

ssh -i ~/path/to/key gmail아이디@공개ip

java 설치

> sudo dnf install java-17-openjdk java-17-openjdk-devel

> java --version
openjdk version "17.0.6" 2023-01-17 LTS
OpenJDK Runtime Environment (Red_Hat-17.0.6.0.10-3.el9) (build 17.0.6+10-LTS)
OpenJDK 64-Bit Server VM (Red_Hat-17.0.6.0.10-3.el9) (build 17.0.6+10-LTS, mixed mode, sharing)

# java home 경로 확인
> ls /usr/lib/jvm/jre-17-openjdk
bin  conf  include  legal  lib  release  tapset

# JAVA_HOME 설정
> echo 'export JAVA_HOME=$(dirname $(dirname $(readlink -f $(which javac))))' | sudo tee -a /etc/profile

> source /etc/profile

> echo $JAVA_HOME
/usr/lib/jvm/java-17-openjdk-17.0.6.0.10-3.el9.x86_64

> ls $JAVA_HOME
bin  conf  include  legal  lib  release  tapset

kafka 설치 및 실행

1) 다운로드

> sudo dnf install -y  wget
> mkdir ~/apps
> cd ~/apps/
> wget https://downloads.apache.org/kafka/3.6.2/kafka_2.12-3.6.2.tgz
> tar -xzvf kafka_2.12-3.6.2.tgz

2) zookepper server 시작

> cd ~/apps/kafka_2.12-3.6.2

# 1) Foreground 실행 (테스트 용으로 zookeeper 로그를 직접 확인)
> bin/zookeeper-server-start.sh config/zookeeper.properties

# 2) Background 실행
> bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
> ps -ef | grep zookeeper

3) kafka server

> cd ~/apps/kafka_2.12-3.6.2

# 1) Foregroud 
> bin/kafka-server-start.sh config/server.properties

# 2) background 실행
> bin/kafka-server-start.sh -daemon config/server.properties

4) kafka data 저장 경로 확인

cd ~/apps/kafka_2.12-3.6.2/cd ~/apps/kafka_2.12-3.6.2

• 여기서 보면 기본적으로 log.dirs=/tmp/kafka-logs 에 저장되어있는걸 알 수 있음
• 변경해줄 필요가 있음

ls -al /tmp/kafka-logs/meta.properties 

• 메타 정보 확인 가능

5) 토픽 만들기

• 일단 서버가 하나이기 때문에 모두 1로 설정
• 이름은 test

> cd ~/apps/kafka_2.12-3.6.2
> bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic test

> bin/kafka-topics.sh --list --bootstrap-server localhost:9092

• topic 확인
  • 파일을 쓰게 되면 내부 topic이 하나 생성됨

bin/kafka-topics.sh --list --bootstrap-server localhost:9092

• 다시 logs 보기
  • topic 이름의 폴더가 하나 생성되는 것을 볼 수 있음.

ls -al /tmp/kafka-logs/

kafka 데이터 흘려보기

• producer
  • broker와 topic을 지정해줘야함.

> cd ~/apps/kafka_2.12-3.6.2
> bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
my message
second message

• consumer
  • broker와 topic을 지정해줘야함.
  • --from-beginning : 처음부터 가지고와

> cd ~/apps/kafka_2.12-3.6.2
> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
my message
second message

• consumer_offsets
  • 파티션이 여러개 생긴것을 볼 수 있음.

 ls /tmp/kafka-logs/

• message log 확인하기

bin/kafka-run-class.sh kafka.tools.DumpLogSegments --deep-iteration --print-data-log --files /tmp/kafka-logs/test-0/00000000000000000000.log

Zookeeper 확인

1) 접속할 zookeeper 정보를 입력하여, zookeeper 내부로 접속

> cd ~/apps/kafka_2.12-3.6.2

> bin/zookeeper-shell.sh localhost:2181
Connecting to localhost:2181
Connecting to localhost:2181
Welcome to ZooKeeper!
JLine support is disabled

WATCHER::

WatchedEvent state:SyncConnected type:None path:null

## Zookeeper에서 제공하는 정보 조회
ls /
[admin, brokers, cluster, config, consumers, controller, controller_epoch, feature, isr_change_notification, latest_producer_id_block, log_dir_event_notification, zookeeper]


## Broker 관련 정보 조회 
ls /brokers
[ids, seqid, topics]


## 현재 broker 목록을 조회 (broker 0번만 존재)
ls /brokers/ids
[0]

## 0번 broker에 대한 정보 확인 
get /brokers/ids/0
{"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},"endpoints":["PLAINTEXT://instance-20240701-162846.asia-northeast3-b.c.beaming-grid-428115-d6.internal:9092"],"jmx_port":-1,"features":{},"host":"instance-20240701-162846.asia-northeast3-b.c.beaming-grid-428115-d6.internal","timestamp":"1719917952131","port":9092,"version":5}


## topic 정보 조회 
ls /brokers/topics
[__consumer_offsets, test]


## topic 상세 정보 조회
get /brokers/topics/test
Node does not exist: /brokers/topics/test

# test topic을 삭제하지 않은 경우에는 아래와 같은 topic 정보가 출력됨
{"removing_replicas":{},"partitions":{"0":[0]},"topic_id":"ZOLr9CLNQjSW70hyIHx6BQ","adding_replicas":{},"version":3}


## controller broker 조회
get /controller
{"version":2,"brokerid":0,"timestamp":"1719917952299","kraftControllerEpoch":-1}