CONA BLOG

# 1d = 86400000ms
$ ./bin/kafka-topics.sh --zookeeper <Zookeeper ip address> --alter --topic \
<Topic name> --config retention.ms=86400000

# 변경되었는지 확인 retention.ms=86400000면 성공
$ ./bin/kafka-topics.sh  --describe --zookeeper <Zookeeper ip address> --topic <Topic name>

참고

https://knight76.tistory.com/entry/kafka%EC%97%90%EC%84%9C-retentionms-%EC%88%98%EC%A0%95%ED%95%98%EB%8A%94-%EB%B0%A9%EB%B2%95

https://github.com/occidere/TIL/issues/13

$ wget http://mirror.navercorp.com/apache/kafka/2.4.0/kafka_2.12-2.4.0.tgz
$ tar -xzf kafka_2.12-2.4.0.tgz
$ cd kafka_2.12-2.4.0


# kafka 폴더 안에서 zookeeper 실행
$ bin/zookeeper-server-start.sh config/zookeeper.properties

# kafka 실행
$ bin/kafka-server-start.sh config/server.properties


# create topic
$ bin/kafka-topics.sh --create --bootstrap-server localhost:9092 \
--replication-factor 1 --partitions 1 --topic <topic name>

# topic 리스트 확인
$ bin/kafka-topics.sh --list --bootstrap-server localhost:9092

참고

https://kafka.apache.org/quickstart

https://soyoung-new-challenge.tistory.com/61

https://sejoung.github.io/2018/09/2018-09-11-kafka_quickstart/#7%EB%8B%A8%EA%B3%84-Use-Kafka-Connect-to-import-export-data

https://12bme.tistory.com/521

Java 설치

$ sudo dnf install java-11-openjdk wget vim

Kafka 다운로드

$ wget http://www-us.apache.org/dist/kafka/2.4.0/kafka_2.13-2.4.0.tgz
$ tar xzf kafka_2.13-2.4.0.tgz
$ mv kafka_2.13-2.4.0 /usr/local/kafka

//root 아닐 경우에는 sudo mv kafka_2.13-2.4.0 /usr/local/kafka

Unit file Setting

Zookeeper

$ vim /etc/systemd/system/zookeeper.service

[Unit]
Description=Apache Zookeeper server
Documentation=http://zookeeper.apache.org
Requires=network.target remote-fs.target
After=network.target remote-fs.target

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

[Install]
WantedBy=multi-user.target

Kafka

$ vim /etc/systemd/system/kafka.service

[Unit]
Description=Apache Kafka Server
Documentation=http://kafka.apache.org/documentation.html
Requires=zookeeper.service

[Service]
Type=simple
Environment="JAVA_HOME=/usr/lib/jvm/jre-11-openjdk"
ExecStart=/usr/local/kafka/bin/kafka-server-start.sh /usr/local/kafka/config/server.properties
ExecStop=/usr/local/kafka/bin/kafka-server-stop.sh

[Install]
WantedBy=multi-user.target

$ systemctl daemon-reload	//마무리

Kafka 실행

$ sudo systemctl start zookeeper	//zookeeper 시작
$ sudo systemctl start kafka		//kafka 시작
$ sudo systemctl status kafka		//kafka active 확인

오류발생

zookeeper 실행 안되는 현상으로 kafka까지 active fail

아래 명령어로 프로세스 조회

$ journalctl -xe

Cloudkarafka

클라우드에서 관리되는 apache kafka 서버로, 카프카의 설정, 실행 및 확장 등을 자동화시킨 서비스를 제공한다.

cloud에 대한 자세한 내용은 아래 아파치 카프카의 공식 홈페이지에서 확인할 수 있다.

https://www.cloudkarafka.com/docs/index.html

cloudkarafka getting start - python

툴은 Visual Studio를 사용

1. 원하는 디렉터리 경로에서 아래 명령어를 입력

git clone https://github.com/CloudKarafka/python-kafka-example.git

2. visual studio에서 해당 프로젝트 오픈

3. 환경변수 설정을 위해서 cloudkarafka의 왼쪽 메뉴바의 [Detail] 클릭

Connection details의 Download를 클릭하여 instance 정보가 담긴 txt 파일을 다운로드

4. 3번의 txt파일의 정보를 가지고, 터미널로 환경변수 설정

~ export CLOUDKARAFKA_BROKERS="rocket-01.srvs.cloudkafka.com:9094,rocket-02.srvs.cloudkafka.com:9094,rocket-03.srvs.cloudkafka.com:9094"
~ export CLOUDKARAFKA_USERNAME="it5v25et"
~ export CLOUDKARAFKA_PASSWORD="xDjmU_mpuzRcoux8pjFkPJAmgutBU0ar"
~ export CLOUDKARAFKA_TOPIC="it5v25et-testTopic"

// TOPIC은 "username-내가정한인스턴스이름"

5. 환경변수 설정을 끝내고 터미널로 consumer.py 실행

~ python consumer.py

6. 또 다른 Terminal을 열고 producer.py 실행

~ python producer.py

7. cloudkarafka에서 producer에서 메시지 적재

메시지 적재가 된 것을 Terminal에서 확인할 수 있다.

Cloudkarafka

클라우드에서 관리되는 apache kafka 서버로, 카프카의 설정, 실행 및 확장 등을 자동화시킨 서비스를 제공한다.

cloud에 대한 자세한 내용은 아래 아파치 카프카의 공식 홈페이지에서 확인할 수 있다.

https://www.cloudkarafka.com/docs/index.html

Cloudkarafka 시작하기

1. 홈페이지에 접속한 뒤 초록색 버튼 클릭 - https://www.cloudkarafka.com/ 

2. 이 포스팅에서 사용할 것은 이 귀여운 DEVELOPER DUCK! 

[Try a Developer Duck]을 클릭 -> 회원가입 진행 -> 로그인

3. 로그인하면 아래와 같은 창이 보이고, Name에는 자신이 원하는 Instance 이름을 입력 -> [Select Region] 클릭

(위의 노란색 알림은 무시해도 상관 없음)

4. data 센터의 위치 선택 -> [review] 클릭

5. [create instance] 클릭

6. instance가 생성된 것을 확인 (testinstance)

7. 방금 만든 인스턴스를 클릭하면 아래와 같이 정보를 확인할 수 있다. 

Consumer & Producer 사용해보기 (message 적재)

이번 포스팅은 producer를 이용해서 메세지를 적재하고, consumer가 소비하는 것을 확인하는 것이 목표!!

1. 왼쪽 메뉴바의 [TOPICS]를 클릭하여 토픽에 대한 정보를 입력 -> [create] 클릭

(간단한 테스트만 진행할 것이기 때문에, partitions와 replicas를 모두 1로 정함)

2. 왼쪽 메뉴바의 [BROWSER] 클릭 -> topic name을 Consumer, Producer에 입력

3. Producer 에 적재하고 싶은 Message를 작성하여 [Produce] 클릭하면 Consumer에서 확인할 수 있다

4. [Consume] 클릭하면 메세지를 소비