섹션4. 카프카 CLI 활용

kafka-topics.sh : 토픽을 명시적으로 만들때 사용

kafka-topics : 토픽관련 옵션

create : 만들겠다

bootstrap-server : 서버를

my-kafka:9092 서버의 주소는 이곳이고

topic : 토픽의 이름은 이거다

 

확인할때 describe 옵션

추가 토픽 hello.kafka2 생성

파티션의 갯수가 10개인것을 확인 (선택옵션을 지정해줘서)

 

토픽 test를 생성. 따로 설정을 안해줘서 기본값으로 지정된모습

 

옵션을 수정할때 사용하는 명령어 alter 를 써서 partitions 갯수를 10개로 늘림

 

만약 alter옵션을 써서 파티션의 갯수를 줄이려한다면?

줄일수는 없다고 에러남

 

 

파티션 갯수를 늘릴수는 있지만, 줄일수는 없다.

파티션 갯수를 늘릴수는 있지만, 줄일수는 없으니 ‘이걸 늘리는게 맞나?’ 에 대한 깊은 고찰이 필요하다.

alter옵션을 통해서 파티션의 갯수를 줄이려고하면 InvalidPartitionsException이 발생함.

 

kafka-configs.sh : 토픽의 옵션수정

configs 에는 현재 세그먼트에 대한 정보만 나와잇음

 

kafka-configs.sh 로 min.insync.replicas 옵션을 추가해줬음

 

configs 옵션에 추가된 것을 확인할수 있음

kafka-console-producer.sh : 콘솔에서 데이터를 넣을수있다

 

메세지 키 설정하지 않음 → 토픽에 넣을때 메세지키는 null

메세지 값만 설정함 ⇒ 라운드 로빈 형식으로 넣어짐

 

그만 보내고 싶으면 ctrl c

property 옵션에서 parse.key로 키를보내겠다 명세하고, key.separator로 직접 지정해줌

k1에는 mymessage value랑 no3가 들어간 모습이 보인다

 

 

 

kafka-console-consumer.sh : 콘솔에서 데이터 조회 가능

 

따로 key 에 대한 지정이 없다보니, value만 나온 상태

 

 

property옵션으로 메세지키-메세지값 형태로 나온것을 확인할수 있음

현재 k1-no3 까지 읽었고(레코드), 이 다음으로 들어온 메세지부터 확인한다

from beginning 옵션으로 “ 가장 처음에 보낸 데이터 중에 “

max messages 1옵션으로 “1개만” 확인한것

partition 옵션으로 파티션 번호 0번을 지정해준상황

근데 파티션 1번을 확인해보니 없다

파티션 2,3번도 없는데

이건 카프카 정책때문에 그런거고

0번에만 다 넣어도 된다고 판단해서 0번에 몰빵한것임

 

 

group옵션으로 컨슈머 그룹 hello-group은

마지막으로 나온 no3까지 커밋했다 라는거임

커밋된 정보는 __consumer_offsets 그룹에 저장됨

 

왼쪽엔 프로듀서, 오른쪽엔 컨슈머

kafka-consumer-groups.sh : 컨슈머들의 상태를 확인하는 목적

프로듀서가 넣었는데, 컨슈머랑 차이가 난다?
→ 지연 발생했다 (Consumer LAG,컨슈머 랙)

LAG : 마지막으로 가져간 오프셋 - 마지막 쌓인 오프셋

아까 생성한 hello-group 컨슈머 그룹이 보임

이런식으로 토픽을 넣는중에 describe로 확인해보면

current offset은 13

logend offset은 20 으로 차이가 나는걸 볼수있음

LAG도 늘어났음 ( 7만큼 ⇒ 7만큼 처리를 못했구나, 7만큼 지연이 생겼다)

그래서 콘솔컨슈머로 토픽을 처리 해 줬고 (왼쪽)

다시 describe 찍어보면 (오른쪽)

랙이랑 다 없어진걸 볼수있음

특정한 위치부터 읽게 시작하는것도 가능함

지금은 reset 오프셋으로 0번부터 읽으라고해서

맨처음에 들어갔던 hello부터 쭉쭉쭉 나오고잇음

그 외 커맨드라인 툴

kafka-producer-peft-test.sh : 프로듀서의 퍼포먼스 측정

kakfa-consumer-peft-test.sh : 컨슈머의 퍼포먼스 측정 (브로커와 컨슈머의 네트워크 체크)

kafka-reassign-partition.sh : 리더 파티션의 쏠림현상 리밸런싱

kafka-delete-record.sh : 레코드를 삭제하는 기능

kafka-dump-log.sh : 상세로그에 대한 확인

토픽을 생성하는 두가지 방법

1. Kafka-topic.sh 에서 create 옵션을 써서 명시적으로 생성하거나
2. auto-create옵션이 True일때, 생성되지않은 토픽에 대해서 데이터를 요청하는 상황에서 자동적으로 생성하는경우

카프카 브로커랑 로컬 커맨드라인툴 버전을 맞춰야하는 이유

버전차이가 나버리면

1. 오류가 났나?
2. 이게 아닌데?
3. 어디서 에러났지?
   의 무한 반복 가능