1.2 Kafka 카프카 빠르게 시작하기 - 카프카 CLI 사용하기

CLI로 토픽 생성해보기

로컬 PC

파티션 개수, 복제 개수 등 여러 옵션들은 모두 디폴트로 설정됨

➜  cd bin

# hello.kafka라는 토픽 생성
➜  ./kafka-topics.sh --create --bootstrap-server my-kafka:9092 --topic hello.kafka

 

카프카 브로커 서버

[ec2-user@ip-172-31-37-178 bin]$ [2023-09-03 12:48:23,724] INFO Creating topic hello.kafka with configuration {} and initial partition assignment Map(0 -> ArrayBuffer(0)) (kafka.zk.AdminZkClient)
[2023-09-03 12:48:23,887] INFO [ReplicaFetcherManager on broker 0] Removed fetcher for partitions Set(hello.kafka-0) (kafka.server.ReplicaFetcherManager)
[2023-09-03 12:48:24,021] INFO [Log partition=hello.kafka-0, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
[2023-09-03 12:48:24,035] INFO [Log partition=hello.kafka-0, dir=/tmp/kafka-logs] Completed load of log with 1 segments, log start offset 0 and log end offset 0 in 84 ms (kafka.log.Log)
[2023-09-03 12:48:24,061] INFO Created log for partition hello.kafka-0 in /tmp/kafka-logs/hello.kafka-0 with properties {compression.type -> producer, message.downconversion.enable -> true, min.insync.replicas -> 1, segment.jitter.ms -> 0, cleanup.policy -> [delete], flush.ms -> 9223372036854775807, segment.bytes -> 1073741824, retention.ms -> 604800000, flush.messages -> 9223372036854775807, message.format.version -> 2.5-IV0, file.delete.delay.ms -> 60000, max.compaction.lag.ms -> 9223372036854775807, max.message.bytes -> 1048588, min.compaction.lag.ms -> 0, message.timestamp.type -> CreateTime, preallocate -> false, min.cleanable.dirty.ratio -> 0.5, index.interval.bytes -> 4096, unclean.leader.election.enable -> false, retention.bytes -> -1, delete.retention.ms -> 86400000, segment.ms -> 604800000, message.timestamp.difference.max.ms -> 9223372036854775807, segment.index.bytes -> 10485760}. (kafka.log.LogManager)
[2023-09-03 12:48:24,064] INFO [Partition hello.kafka-0 broker=0] No checkpointed highwatermark is found for partition hello.kafka-0 (kafka.cluster.Partition)
[2023-09-03 12:48:24,066] INFO [Partition hello.kafka-0 broker=0] Log loaded for partition hello.kafka-0 with initial high watermark 0 (kafka.cluster.Partition)
[2023-09-03 12:48:24,070] INFO [Partition hello.kafka-0 broker=0] hello.kafka-0 starts at leader epoch 0 from offset 0 with high watermark 0. Previous leader epoch was -1. (kafka.cluster.Partition)
[2023-09-03 12:48:42,051] INFO [GroupMetadataManager brokerId=0] Removed 0 expired offsets in 0 milliseconds. (kafka.coordinator.group.GroupMetadataManager)

 

토픽 리스트 조회

# 원격 카프카 서버의 토픽 리스트 확인하기
➜  ./kafka-topics.sh --bootstrap-server my-kafka:9092 --list
hello.kafka

 

토픽 상세 조회

기본 옵션으로 만들어서 파티션, Replication 개수가 1개구나

➜ ./kafka-topics.sh --bootstrap-server my-kafka:9092 --describe --topic hello.kafka
Topic: hello.kafka	PartitionCount: 1	ReplicationFactor: 1	Configs: segment.bytes=1073741824
	Topic: hello.kafka	Partition: 0	Leader: 0	Replicas: 0	Isr: 0

 

파티션이 3개인 새로운 토픽 생성 후 확인

# 파티션 개수가 3개인 토픽 생성
➜  ./kafka-topics.sh --create --bootstrap-server my-kafka:9092 --partitions 3 --topic bonjour.kafka
Created topic bonjour.kafka.

# 토픽 상세 확인
➜  bin ./kafka-topics.sh --bootstrap-server my-kafka:9092 --describe --topic bonjour.kafka             
Topic: bonjour.kafka	PartitionCount: 3	ReplicationFactor: 1	Configs: segment.bytes=1073741824
	Topic: bonjour.kafka	Partition: 0	Leader: 0	Replicas: 0	Isr: 0
	Topic: bonjour.kafka	Partition: 1	Leader: 0	Replicas: 0	Isr: 0
	Topic: bonjour.kafka	Partition: 2	Leader: 0	Replicas: 0	Isr: 0

 

프로듀서를 실행해서 토픽에 데이터(레코드) 넣어보기

kafka-console-producer.sh를 이용해 새로운 레코드를 토픽에 넣어보자

(프로듀서 프로세스를 종료하기 위해서는 `Ctrl+c`)

➜  ./kafka-console-producer.sh --bootstrap-server my-kafka:9092 --topic hello.kafka
>hello kafka!
>I'm learning Kafka
>hey
>Can you hear me, server?

 

컨슈머를 실행해서 토픽의 레코드 읽기

새로운 터미널 윈도우를 켜서 다른 클라이언트로 컨슈머를 실행해본다.

➜  ./kafka-console-consumer.sh --bootstrap-server my-kafka:9092 --topic hello.kafka --from-beginning
hello kafka!
I'm learning Kafka
hey
Can you hear me, server?

# 컨슈머 그룹을 생성해서 컨슈밍
➜  ./kafka-consumer-groups.sh --bootstrap-server my-kafka:9092 --group hello-group           

# 컨슈머 그룹 상세 확인
➜  ./kafka-consumer-groups.sh --bootstrap-server my-kafka:9092 --group hello-group