Apache Kafka 설치

0. 가상환경 설정
$ sudo apt install python3.9
$ mkdir kafka
$ cd kafka
$ virtualenv --python=python3.9 kafka-env
$ source ./kafka-env/bin/activate
1. https://kafka.apache.org/downloads 에서 Binary downloads: Scala 2.13 버전 다운로드 후 압축 해제
$ tar xvf kafka_2.13-3.2.1.tgz
$ ln -s kafka_2.13-3.2.1 kafka  # 심볼릭 링크 만들기

image

image

image

bin은 kafka 를 실행하기 위한 shell script가 들어있고, config는 설정 파일 리스트들이 들어있다.

2. zookeeper 띄우기

kafka를 띄우기 위해서 zookeeper를 띄워야 한다. config/zookeeper.properties를 살펴보면 zookeeper는 /tmp/zookeeper 에 데이터를 저장하고, 기본적으로 2181 포트를 사용한다.

image

zookeeper는 아래 명령어로 띄우면 되는데, INFO binding to port 0.0.0.0/0.0.0.0:2181 라는 log가 나오면 정상적으로 잘 작동한 것이다.

$ sudo apt-get install openjdk-11-jdk
$ bin/zookeeper-server-start.sh config/zookeeper.properties

image

zookeeper에 접속이 잘 되는 것을 확인한다. ls 명령어도 잘 작동한다.

$ bin/zookeeper-shell.sh localhost:2181

image

3. kafka broker 띄우기

config/server.properties 파일의 listeners=PLAINTEXT://:9092 를 listeners=PLAINTEXT://127.0.0.1:9092로 변경해준다.

$ vim config/server.properties

image

broker를 띄우는 커맨드는 아래와 같다. INFO [KafkaServer id=0] started log가 뜨면 정상적으로 실행된 것이다.

$ bin/kafka-server-start.sh config/server.properties

image

test를 위해 kafka 토픽을 하나 생성해보자. list를 통해 확인해봐도 생성된 것을 확인할 수 있다. 토픽의 이름은 최대 249자까지로, 공백을 허용하지 않고, _와 .은 내부에서 사용하는 metric 충돌을 방지하기 위해 사용하지 않는 것이 좋다.

$ bin/kafka-topics.sh --create --topic <topic name> --bootstrap-server <kafka server endpoint> 
$ bin/kafka-topics.sh --list --bootstrap-server <kafka server endpoint>

image

메세지 produce, consume

이제 메세지를 produce하고 consume 해본다.

$ bin/kafka-console-producer.sh --bootstrap-server <kafka server endpoint> --topic <topic name>  # producer interactive shell 
$ bin/kafka-console-consumer.sh --bootstrap-server <kafka server endpoint> --topic <topic name>  # consumer interactive shell

producer를 키고 메세지를 보낸 후에 consumer를 키면 consumer는 아무 메세지를 받아오지 않는다.

image

consumer를 킨 상태에서 producer interactive shell에서 메세지를 입력하면 consumer가 메세지를 잘 받아오는 것을 확인할수 있다.

image

만약에 consumer로 producer 가 만든 메세지를 처음부터 보고 싶을 때는 –from-beginning 옵션을 주면 된다.

$ bin/kafka-console-consumer.sh --bootstrap-server <kafka server endpoint> --topic <topic name> --from-beginning

시험 삼아 입력했던 모든 메세지들을 모두 consuming 한것을 볼 수 있다.

image

다시 zookeeper shell에 접속해서 ls를 쳐보면 처음에는 [zookeeper] 만 존재했지만 지금은 kafka broker가 뜨면서 많은 것들이 생겼다.

image

broker의 topic에는 아까 만들었던 mytopic이 있고, __consumer_offsets이 자동으로 생겼다. kafka는 consumer group이 읽어간 offset 정보를 저장하고 있는데, 그 offset 정보를 kafka의 hidden topic인 __consumer_offsets에 저장한다.

image