Kafka 주요 명령어 (window + git bash 기준)

**2장**

<[로컬] Git Bash에서 수행시, ll 명령어 활성화>
echo "alias ll='ls --color=auto -alF'" >> ~/.bashrc
source ~/.bashrc

<[로컬] 권한 변경 및 확인>
chmod 400 kafka-test-1-key-pair.pem
ll kafka-test-1-key-pair.pem

<[로컬] ssh 접속>
ssh -i "D:\YGW\학교\4학년_2학기\3 졸업프로젝트2(종합설계)\kafka\kafka-test-1-key-pair.pem" ec2-user@34.228.41.218

<[인스턴스] JDK 설치>
sudo yum install -y java-1.8.0-openjdk-devel.x86_64

<[인스턴스] 카프카 설치>
wget https://archive.apache.org/dist/kafka/2.5.0/kafka_2.12-2.5.0.tgz

<[인스턴스] 카프카 압축 풀기>
tar xvf kafka_2.12-2.5.0.tgz
cd kafka_2.12-2.5.0

<[인스턴스] t2.micro라는 1GB 메모리 환경이므로, 힙 메모리 사이즈 변경 및 확인>
* 일회성 (터미널 종료 시 만료)
export KAFKA_HEAP_OPTS="-Xmx400m -Xms400m"
echo $KAFKA_HEAP_OPTS
* 영구성
vi ~/.bashrc
# User specific aliases and functions에
export KAFKA_HEAP_OPTS="-Xmx400m -Xms400m"
입력
source ~/.bashrc (수정한 값 반영)
echo $KAFKA_HEAP_OPTS

<[인스턴스] 카프카 브로커 실행 옵션 설정>
vi config/server.properties
advertised.listeners=PLAINTEXT://34.228.41.218:9092 (주석 해제 및 인스턴스 IP주소 입력)

<[인스턴스] 주키퍼 실행 및 확인>
bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
jps -vm (m옵션은 main 메서드에 전달된 인자 확인 / v옵션은 jvm에 전달된 인자 확인)

<[인스턴스] 브로커 실행 및 확인>
bin/kafka-server-start.sh -daemon config/server.properties
jps -m
tail -f logs/server.log

<[로컬] 카프카 툴 설치 : 오류 발생 시 cmd로 실행하거나 git bash 최신 버전으로 재설치>
curl https://archive.apache.org/dist/kafka/2.5.0/kafka_2.12-2.5.0.tgz --output kafka.tgz
tar -xvf kafka.tgz (압축 해제)
cd kafka_2.12-2.5.0/bin
* 원격 AWS 가상 머신의 카프카 서버에 접속하여 정보 확인
./kafka-broker-api-versions.sh --bootstrap-server 34.228.41.218:9092
(윈도우의 경우) ./windows/kafka-broker-api-versions.bat --bootstrap-server 34.228.41.218:9092
* jdk 오류 발생 시
vi ~/.bashrc 에서
export JAVA_HOME='/c/Program Files\Java\jdk1.8.0_281'
export PATH=$JAVA_HOME/bin:$PATH
입력
source ~/.bashrc (수정한 값 반영)

<[로컬] 테스트 편의를 위한 hosts 설정 : bash를 관리자 권한으로 켤 것>
vi c:/Windows/System32/drivers/etc/hosts
34.228.41.218 my-kafka 추가

<[로컬] 토픽 생성>
cd windows
./kafka-topics.bat --create --bootstrap-server my-kafka:9092 --topic hello.kafka
./kafka-topics.bat --create --bootstrap-server my-kafka:9092 --partitions 3 --replication-factor 1 --config 
retention.ms=172800000 --topic hello.kafka.2

<[로컬] 토픽 리스트 조회>
./kafka-topics.bat --bootstrap-server my-kafka:9092 --list

<[로컬] 토픽 상세 조회>
./kafka-topics.bat --bootstrap-server my-kafka:9092 --describe --topic hello.kafka.2

<[로컬] 토픽 옵션 수정>
./kafka-topics.bat --bootstrap-server my-kafka:9092 \
--topic hello.kafka \
--alter --partitions 4
./kafka-topics.bat --bootstrap-server my-kafka:9092 \
--topic hello.kafka \
--describe

./kafka-configs.bat --bootstrap-server my-kafka:9092 \
--entity-type topics \
--entity-name hello.kafka \
--alter --add-config retention.ms=86400000
./kafka-configs.bat --bootstrap-server my-kafka:9092 \
--entity-type topics \
--entity-name hello.kafka \
--describe

<[로컬] console producer (without key)>
./kafka-console-producer.bat --bootstrap-server my-kafka:9092 \
--topic hello.kafka

<[로컬] console producer (with key)>
./kafka-console-producer.bat --bootstrap-server my-kafka:9092 \
--topic hello.kafka \
--property "parse.key=true" \
--property "key.separator=:"

<[로컬] console consumer (without key)>
./kafka-console-consumer.bat --bootstrap-server my-kafka:9092 \
--topic hello.kafka \
--from-beginning

<[로컬] console consumer (with key)>
./kafka-console-consumer.bat --bootstrap-server my-kafka:9092 \
--topic hello.kafka \
--property print.key=true \
--property key.separator="-" \
--group hello-group \
--from-beginning

<[로컬] consumer group>
* 그룹 목록 확인
./kafka-consumer-groups.bat --bootstrap-server my-kafka:9092 --list
* 그룹 상세 확인
./kafka-consumer-groups.bat --bootstrap-server my-kafka:9092 \
--group hello-group \
--describe

<[로컬] verify>
./kafka-verifiable-producer.sh --bootstrap-server my-kafka:9092 \
--max-messages 10 \
--topic verify-test
./kafka-verifiable-consumer.sh --bootstrap-server my-kafka:9092 \
--topic verify-test \
--group-id test-group

----------------------------------------------

**3장**

<[로컬] zookeeper>
./zookeeper-shell.bat my-kafka:2181
-> ls /
-> get /brokers/ids/0
-> get /controller
-> ls /brokers/topics

# 3.4.1. 프로듀서 API
<[로컬] 토픽 생성>
./kafka-topics.bat --bootstrap-server my-kafka:9092 --create \
--topic test \
--partitions 3

<[로컬] 토픽 내용 확인>
./kafka-console-consumer.bat --bootstrap-server my-kafka:9092 \
--topic test \
--from-beginning

./kafka-console-consumer.bat --bootstrap-server my-kafka:9092 \
--topic test \
--property print.key=true \
--property key.separator="-" \
--from-beginning

# 3.4.2. 컨슈머 API
<[로컬] polling 테스트를 위한 produce>
./kafka-console-producer.bat --bootstrap-server my-kafka:9092 --topic test
> testMessage

# 3.5.1. 스트림즈 DSL
<[로컬] stream_log 토픽 생성 후 produce & consume>
./kafka-topics.bat --create \
--bootstrap-server my-kafka:9092 \
--partitions 3 \
--topic stream_log
./kafka-console-producer.bat --bootstrap-server my-kafka:9092 \
--topic stream_log
>hello
>kafka
>stream
./kafka-console-consumer.bat --bootstrap-server my-kafka:9092 \
--topic stream_log_copy --from-beginning