[Kafka] Ubuntu 20.04: Kafka 2.13-3.1.0 install

우분투에서 카프카 설치

kafka사용 시 사용할 사용자 추가

카프카가 네트워크 요청을 처리할 수 있기 때문에,
누군가가 카프카 서버를 손상 시킬 시 우분투 손상을 최소화 하기 위해 kafka 전용 사용자를 새로 만들어 사용하는 것이 좋다

$ sudo adduser kafka

• kafka 사용자 추가

$ sudo adduser kafka sudo

• kafka 사용자를 sudo 그룹에 추가

$ su -l kafka

• kafka 사용자에 로그인

kafka binary 파일 다운로드 및 압축 풀기

$ mkdir ~/Downloads

• 파일을 저장할 Downloads 폴더 생성

$ curl "https://downloads.apache.org/kafka/3.1.0/kafka_2.13-3.1.0.tgz" -o ~/Downloads/kafka.tgz

• kafka binary 파일 다운로드

$ mkdir ~/kafka
$ cd kafka

• kafka를 설치할 kafka 폴더 생성 및 이동

$ tar -xvzf ~/Downloads/kafka.tgz --strip 1

• tar 명령어로 kafka.tgz 파일 압축 풀기
• -x tar 파일 풀기
• -v 처리 과정 표시
• -z gzip 압축 풀기
• -f 파일 지정

---

gzip: stdin: not in gzip format | tar: Error is not recoverable: exiting now 에러

.taz 파일을 푸는 과정에서 아래와 같은 에러가 발생했다.

gzip: stdin: not in gzip format
tar: Child returned status 1
tar: Error is not recoverable: exiting now

• 파일이 gzip format이 아니다.

파일 형식 확인

file ~/Downloads/kafka.tgz

• kafka.tgz 파일의 파일 형식을 확인
• HTML document, ASCII text
• ...일리가 없는데, 애초에 파일 다운부터 문제가 있었던 것 같다.

파일 재다운로드

$ rm kafka.tgz

• 기존에 다운로드 했던 binary 파일을 삭제하고

• https://kafka.apache.org/downloads 에 접속해
• Scala 2.13 - kafka_2.13-3.1.0.tgz (asc, sha512)의 파일 링크를 복사해 다시 다운로드 명령어에 입력해줬다.

$ curl "https://archive.apache.org/dist/kafka/3.1.0/kafka_2.13-3.1.0.tgz" -o ~/Downloads/kafka.tgz

• 다시 해주니 다운로드 시간이 꽤 걸린다. 어쩐지 아까 다운로드가 너무 금방 되더라...

파일 압축 해제

$ tar -xvzf ~/Downloads/kafka.tgz --strip 1

• 다시 압축 해제를 실행하니 잘 됐다! 에러 해결

---

카프카 서버 구성

카프카의 초기 설정은 topic 삭제가 불가능하도록 되어있다.
이러한 초기 설정을 수정해주자

$ vi ~/kafka/config/server.properties

• 파일 맨 아래에 delete.topic.enable = true 추가
• topic을 삭제 가능하도록 설정
  
• log.dir 경로 변경
• 카프카 로그 저장하는 경로를 변경

systemd unit file 생성 및 카프카 서버 실행

zookeeper를 위한 유닛 파일 생성

$ sudo vi /etc/systemd/system/zookeeper.service

[Unit]
Requires=network.target remote-fs.target
After=network.target remote-fs.target

[Service]
Type=simple
User=kafka
ExecStart=/home/kafka/kafka/bin/zookeeper-server-start.sh /home/kafka/kafka/config/zookeeper.properties
ExecStop=/home/kafka/kafka/bin/zookeeper-server-stop.sh
Restart=on-abnormal

[Install]
WantedBy=multi-user.target

• vi로 파일을 생성하고 열어서 위의 내용을 입력해준다.[Unit]
• Zookeeper가 시작하기 전에 네트워킹 및 파일 시스템을 준비해야 한다고 지정[Service]
• systemd가 zookeeper-server-start.sh 및 zookeeper-server-stop을 사용하도록 지정
• sh 셸 파일을 사용하여 서비스를 시작 및 중지할 수 있음
• 비정상적으로 종료되는 경우 Zookeeper를 다시 시작해야 함을 지정

kafka systemd service file 생성

$ sudo vi /etc/systemd/system/kafka.service

[Unit]
Requires=zookeeper.service
After=zookeeper.service

[Service]
Type=simple
User=kafka
ExecStart=/bin/sh -c '/home/kafka/kafka/bin/kafka-server-start.sh /home/kafka/kafka/config/server.properties > /home/kafka/kafka/kafka.log 2>&1'
ExecStop=/home/kafka/kafka/bin/kafka-server-stop.sh
Restart=on-abnormal

[Install]
WantedBy=multi-user.target

• vi로 파일을 생성하고 열어서 위의 내용을 입력해준다.[Unit]
• 해당 유닛 파일이 zookeeper.service에 의존함을 지정
• 카프카 서비스가 시작할 때 자동으로 zookeeper 실행[Service]
• systemd가 kafka-server-start.sh 및 kafka-server-stop.sh 를 사용하도록 지정
• sh 셸 파일을 사용하여 서비스를 시작 및 중지할 수 있음
• 비정상적으로 종료되는 경우 카프카를 다시 시작해야 함을 지정

$ sudo systemctl start kafka

• 카프카 실행

$ sudo systemctl status kafka

• 카프카 상태 확인
• 정상 실행!
• 이제 포트 9092에서 Kafka 서버가 수신된다.

$ sudo systemctl enable zookeeper
$ sudo systemctl enable kafka

• 서버 재부팅시 카프카 서비스를 자동으로 시작하도록 설정

카프카 서버 테스트

✅ 카프카에서 메시지를 게시하려면 아래 두 가지가 필요하다

• Producer: records와 data를 topic에 넣어주는 역할
• Consumer: topic에서 message와 data를 읽는 역할

$ ~/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic TutorialTopic

• TutorialTopic 이라는 topic 생성
• kafka-console-producer.sh 스크립트로 producer 생성 (Kafka server’s hostname, a port, and a topic...)

$ echo "Hello, World" | ~/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic TutorialTopic > /dev/null

• producer 쉘로 TutorialTopic topic에 "Hello, World" publish
• kafka-console-consumer.sh 스크립트로 consumer 생성 (ZooKeeper server’s hostname and port, topic name...)

$ ~/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic TutorialTopic --from-beginning

• Output "Hello, World" 확인
• consumer 쉘로 TutorialTopic topic으로부터 message 가져옴
• --from-beginning: consumer가 시작되기 전에 publish된 메세지를 사용할 수 있도록하는 flag

✅ 메세지를 수신 받아도 script는 계속 run되고 있음
새로 data가 publish되면 consumer가 바로 받아오기위해 대기함
터미널을 새로 열어

$ echo "Hello World from Sammy at DigitalOcean!" | ~/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic TutorialTopic > /dev/null

• producer로 "Hello World from Sammy at DigitalOcean!"를 topic에 publish 해보면

• consumer가 바로 "Hello World from Sammy at DigitalOcean!"을 받아옴을 확인

Ctrl + C로 종료

---

References

https://www.digitalocean.com/community/tutorials/how-to-install-apache-kafka-on-ubuntu-20-04
https://askubuntu.com/questions/909994/tar-gz-file-downloaded-as-html-document-ascii-text
https://linuxhint.com/solve-gzip-stdin-not-gzip-format-error/