Kafka 설치 & 설정 & 실행 (Windows WSL 환경)

1. git 설치
https://git-scm.com/download/win

Redirecting…

 

2. WSL 환경 설치
Windows Subsystem for Linux는 리눅스처럼 bash쉘을 활용하는 것을 뜻합니다. 
https://learn.microsoft.com/ko-kr/windows/wsl/about

Linux용 Windows 하위 시스템란?

 

 

기본적으로 Windows 기능 켜기/끄기에서 다음 내역을 활성화 합니다.
1) Linux용 Windows 하위 시스템
2) Windows 하이퍼바이저 플랫폼
3) 가상 머신 플랫폼

 

1) Ubuntu를 마이크로소프트 스토어에서 설치
https://learn.microsoft.com/ko-kr/windows/wsl/install-manual#step-4---download-the-linux-kernel-update-package

이전 버전의 WSL에 대한 수동 설치 단계

 

2) JDK 1.8 설치
터미널을 실행하고 bash를 입력하여 쉘을 변경합니다.

$ bash
이후에 아래 명령어를 실행하여 openjdk-8-jdk를 설치합니다.
$ sudo apt update
$ sudo apt install openjdk-8-jdk

설치가 완료되면 아래 명령어를 쳐서 openjdk-8-jdk가 잘 설치되었는지 확인합니다.
$ java -version

3. 카프카 바이너리 파일 다운로드 및 압축 풀기
https://kafka.apache.org/downloads

Apache Kafka

2.X.X 버전의 Scala 2.13을 다운로드

$ cd Downloads
$ tar -xvf kafka_2.13-2.7.0.tgz
$ ls
$ cd /mnt/c/Kafka/kafka_2.13-2.7.0
kafka_2.13-2.7.0 을 c:/Kafka 로 이동

 

1) Heap Size 설정 제거
export KAFKA_HEAP_OPTS="-Xmx400m -Xms400m"

$ sysctl -w vm.max_map_count=262144
$ cat /proc/[kafka-pid]/maps | wc -l

 

2) 설정  : config/server.properties

  broker.id=0
  listeners=PLAINTEXT:// localhost:9092
  advertised.listeners=PLAINTEXT://localhost:9092

  
3) 주키퍼, 카프카 브로커 실행하기
주키퍼는 다음 명령어로 실행

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

카프카 브로커를 다음 명령어로 실행

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

4) 로그확인

$ tail -f logs/*

5) 카프카 명령어 테스트하기
새로운 터미널 탭을 열고 카프카 바이너리 위치로 이동하여 다음 코드를 실행

$ bin/kafka-topics.sh --bootstrap-server localhost:9092 --topic test --create

실행이 잘 되었다면 다음과 같이 토픽 이름을 확인

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

6) 생성된 Topic에 Message 보내기 (Producer)

$ bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test

7) Topic에서 Message 가져오기 (Consumer)

$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test
$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
—from-beginning 옵션을 추가하면 기존의 발행한 메세지도 모두 가져옵니다.

8) Consumer Group 으로 Message 가져오기

$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test -group testgroup --from-beginning
$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list
$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group testgroup --describe

 

PARTITION : 파트션 갯수

CURRENT-OFFSET :  컨슈머 오프셋 (by testgroup)

LOG-END-OFFSET :  토픽의 마지막 오프셋

LAG : 컨슈머 랙 = CURRENT-OFFSET - LOG-END-OFFSET

 

9) OFFSET RESET

$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group testgroup --topic test --reset-offsets --to-earliest --execute
$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group testgroup --topic test:1 --reset-offsets --to-offset 10 --execute

to-earliest : 가장 먼저 생성된 

특정파티션 : test:1

to-offset 10 : 10 번 offset 부터