[AWS] : building kafka cluster

kafka / zookeeper cluster 구축 내용

각 hostname이 worker1, worker2, worker3인 AWS instance 3개를 통해 cluster를 구축한다.

zookeeper : 분산 시스템을 위한 코디네이터로 kafka의 동작은 zookeeper에 의해 관리된다. 별도로 zookeeper를 구축해도 되지만, 해당 프로젝트 에서는 kafka 설치 시 들어있는 zookeeper를 사용한다.

zookeeper download : apache-zookeeper-3.7.0
- wget을 명령어를 통해 각 AWS instance의 /usr/local에 다운받는다.
- tar zxf를 통해 압축을 풀어준다.
- 압축을 푼 zookeeper-3.7.0 폴더에symbolic link를 생성해준다.
  - 버전 정보가 있는 폴더는 symbolic link를 생성하여 추후 버전이 바뀌는 경우 배포 스크립트 등에 설정된 경로를 모두 변경하지 않아도 되도록 하는 것이 좋다.
  - symbolic link는 ln -s를 통해 생성한다.
zookeeper 설정
- zookeeper가 지노드의 복사본인 스냅샤과 트랜잭션 로그들을 저장할 /zoo 경로를 생성한다.
  - 원하는 경로를 따로 지정해도 된다.
  - 이후 zookeeper node를 구분하기 위한 id를 /zoo하단에 myid라는 파일을 만들고 각 숫자를 입력한다.
    - worker1 -> myid : 1
    - worker2 -> myid : 2
    - worker3 -> myid : 3
- /usr/local/zookeeper/conf아래에 zookeeper의 환경설정 파일 zoo.cfg를 생성한다.
  - zoo.cfg에서 필요한 내용 구성을 아래와 같이 기술한다.
    - zookeeper Admin docs 참조
zookeeper 실행 : zookeeper에서 제공하는 zookeeper/bin/zkServer.sh start와 같이 start하거나 stop할 수 있다.
- systemd 등록 : 여러 프로세스를 systemd에 등록하여 운영하도록 한다.
  - /etc/systemd/system에 zookeeper-server.service라는 파일명으로 zookeeper용 systemd파일을 아래와 같이 생성.
  - systemd의 파일을 새로 만들거나 수정하였을 경우에는 systemd 재시작을 위해 systemctl daemon-reload를 입력해야 하며 이후 systemctl명령어를 통해 zookeeper-server.service를 start하거나 stop하는 등의 명령어로 worker1,worker2,worker3에서 사용할 수 있다.
    - zookeeper가 정상적으로 실행 되었음은 systemd에 등록하였기 때문에 systemctl명령어를 통해 알아낼 수 있다.
  - daemon process란?
  - systemctl 명령어 사용법

kafka : 기본적으로 pub/sub model을 구현한 분산 메시징 시스템이다. kafka는 수평적인 확장을 위해 cluster를 구성하며 broker의 clustering을 위해 zookeeper를 통해 관리한다.

kafka download : kafka-2.8.0 / scala 2.12
- wget을 명령어를 통해 각 AWS instance의 /usr/local에 다운받는다.
- tar zxf를 통해 압축을 풀어준다.
- 압축을 푼 kafka_2.12-2.8.0 폴더에 대한 symbolic link를 zookeeper와 같이 생성해준다.
  - 버전 정보가 있는 폴더는 symbolic link를 생성하여 추후 버전이 바뀌는 경우 배포 스크립트 등에 설정된 경로를 모두 변경하지 않아도 되도록 하는 것이 좋다.
  - symbolic link는 ln -s를 통해 생성한다.
kafka 설정 : server별 broker id, kafka 저장 디렉토리, zookeeper 정보 등을 설정
- kafka는 일반 message queue와는 다르게 pub/sub구조이므로 consumer가 메시지를 가져가더라도 저장된 data를 임시 보관하는 기능이 존재하며 분산 저장을 위한 여러 저장 디렉토리를 생성해도된다.
  - 원하는 경로 및 디렉토리 명을 따로 지정해도 된다.
  - 이 프로젝트 에서는 /kafka1과 /kafka2를 생성한다.
- zookeeper server와의 통신 확인 : kafka broker server들은 zookeeper server와의 통신이 되어야하기 때문에, 방화벽으로 포트 접근이 제한된 환경인지 아닌지 등을nc 명령어를 통해 확인해봐야 한다.
  - nc -v hostname 2181을 통해 각 instance들에 대해 확인해본다.
- /usr/local/kafka/config아래의 server.properties파일에 위에서 언급했던 broker.id, 디렉토리, zookeeper 정보를 아래와 같이 설정한다.
  - 환경설정 내용을 적용시키려면 수정 후 반드시 kafka cluster를 재시작 해야 한다.
kafka 실행 : kafka에서 제공하는 kafka 설치경로 하위 경로의 bin에 kafka-server-start.sh를 kafka 환경설정 파일 /usr/local/kafka/config/server.properties를 옵션으로 주어 실행할 수 있으며 background실행 시 &나 -daemon option을 추가할 수 있다. stop도 마찬가지로 bin에 kafka-server-stop.sh를 실행시킨다.
- systemd 등록 : 여러 프로세스를 systemd에 등록하여 운영하도록 한다.
  - /etc/systemd/system에 kafka-server.service라는 파일명으로 kafka용 systemd파일을 아래와 같이 생성.
  - systemd의 파일을 새로 만들거나 수정하였을 경우에는 systemd 재시작을 위해 systemctl daemon-reload를 입력해야 하며 이후 systemctl명령어를 통해 kafka-server.service를 start하거나 stop하는 등의 명령어로 worker1,worker2,worker3에서 사용할 수 있다.
    - kafka가 정상적으로 실행 되었음은 systemd에 등록하였기 때문에 systemctl명령어를 통해 알아낼 수 있다.
  - systemctl 명령어 사용법
kafka 상태 확인: zookeeper와 kafka의 상태가 모든 instance에 대해 systemd를 통한 실행 시 active(running)상태인 경우 kafka와 zookeeper연결 확인을 위한 로그 확인
- TCP 포트 확인 : kafka의 기본 TCP 포트는 9092, zookeeper의 기본 TCP 포트는 2181이므로 netstat명령어를 통해 상태를 확인해본다.
  - netstat -ntlp | grep {portNum} 이후 _LISTEN_하고 있는지를 kafka와 zookeeper 기본 포트에 대해 모든 instance를 확인.

참고

kafka는 cluster를 3대로 구성했을 때, 2대가 다운되어도 서비스를 제공할 수 있는 반면 zookeeper는 3대의 앙상블 중 2대(과반수 이상)가 다운되면 장애가 발생한다. 이와 같이 zookeeper의 장애 상황으로 코디네이션 역할을 하지 못하는 경우에는 kafka cluster에 문제가 없더라도 kafka 역시 장애 상황에 놓이기 때문에 되도록이면 zookeeper와 kafka는 동일한 서버가 아닌 별도의 서버로 구성하는 것이 좋다.

해당 프로젝트에서는 동일 서버에 다른 port로 zookeeper와 kafka를 구축한다.
- kafka : 9092
- zookeeper : 2181

zoo.cfg

/usr/local/zookeeper/conf 경로에 위치

tickTime=2000          # zookeeper가 사용하는 시간 측정 단위
initLimit=10               # follower와 leader가 초기에 연결하는 시간에 대한 timeout
syncLimit=5              # follower와 leader가 동기화 하는 시간에 대한 timeout
dataDir=/data           # zookeeper의 transaction log와 snapshot이 저장되는 데이터 저장 경로
clientPort=2181        # zookeeper 사용 TCP port

# zookeeper 앙상블 구성을 위한 server 설정
# 위에서 지정한 myid의 번호를 통해 server.myid와 같이 구성하며 2888과 3888은 기본 포트이다.
server.1=worker1:2888:3888
server.2=worker2:2888:3888
server.3=worker3:2888:3888

zookeeper-server.service

/etc/systemd/system 경로에 위치

[Unit]
Description=zookeeper-server        # `systemctl status` 명령어에 표시
After=network.target                        

[Service]
Type=forking                # 'ExecStart`에 영향을 주는 unit process가 시작됨
User=root
Group=root
SyslogIdentifier=zookeeper-server
WorkingDirectory=/usr/local/zookeeper            # 실행 process의 working directory지정
Restart=always                # always로 설정한 경우, systemctl 명령어로 인한 중지를 제외하고 process가 종료된 후 항상 재시작 함
RestartSec=0s                  # Restart를 몇 초에 실행할지 지정
ExecStart=/usr/local/zookeeper/bin/zkServer.sh start            # Service가 시작될 때 실행할 명령어 또는 스크립트를 설정
ExecStop=/usr/local/zookeeper/bin/zkServer.sh stop            # Service가 정지될 때 실행할 명령어 또는 스크립트를 설정

zookeeper 실행 장애 발생

worker1,worker2,worker3에 대해 위와 같은 zookeeper 설정 및 실행 진행 시 /usr/local/zookeeper/bin/zkServer.sh start zookeeper server FAILED TO START.. 및 systemctl start zookeeper-server.service 명령 또한 실패

zookeeper ensemble fails to start 원인 및 ../zookeeper/logs/XXX.out확인 결과, org.apache.zookeeper.server.quorum.QuorumPeerMain 에 대한 ClassNotFoundException 발생 확인.

zookeeper-3.7.0.bin.tar.gz로 다시 다운로드하여 해결 가능 (참조)
- zookeeper-server.service는 /etc/systemd/system경로에, myid는 /data 경로에 이미 생성되어있음.
- zookeeper-3.7.0.bin.tar.gz을 기존과 같이 /usr/local에 다운로드 하고 zookeeper symbolic link를 재 지정 후 zookeeper/conf내부에 zoo.cfg를 다시 작성]
- /usr/local/경로의 zookeeper symbolic link를 변경하기 위해서는 ln -Tfs {symbolic link지정 경로} {symbolic link 이름}을 통해 변경할 수 있다.
- 모든 zookeeper ensemble 구축 instance에 대해 진행

kafka server.properties

/usr/local/kafka/config/ 경로에 위치

kafka host 이름과 broker id 설정
- worker1 -> broker.id=1
- worker2 -> broker.id=2
- worker3 -> broker.id=3
디렉토리 지정
- log.dirs항목에 생성해 두었던 /kafka1,/kafka2를 기술
zookeeper 정보 수정
- zookeeper.connect항목에 zookeeper 구축 hostname:port를 차레로 입력한다.

자세한 내용은 broker configs참조

kafka-server.service

/etc/systemd/system 경로에 위치

[Unit]
Description=kafka-server        # `systemctl status` 명령어에 표시
After=network.target                        

[Service]
Type=simple                # 'ExecStart`에 영향을 주는 unit process가 시작됨
User=root
Group=root
SyslogIdentifier=kafka-server
WorkingDirectory=/usr/local/kafka            # 실행 process의 working directory지정
Restart=no                # 항상 재시작을 허용하지 않음
RestartSec=0s                  # Restart를 몇 초에 실행할지 지정
ExecStart=/usr/local/kafka/bin/kafka-server-start.sh /usr/local/kafka/config/server.properties            # Service가 시작될 때 실행할 명령어 또는 스크립트를 설정
ExecStop=/usr/local/kafka/bin/kafka-server-stop.sh            # Service가 정지될 때 실행할 명령어 또는 스크립트를 설정

yeollow / return-home-safely