[Kafka] - 카프카 CLI 명령어 핵심 정리

 

Apache Kafka를 효과적으로 관리하고 운영하기 위해서는 CLI(Command Line Interface)도구에 대한 이해가 필수다.

카프카는 다양한 쉘 스크립트를 제공하여 토픽 관리, 메세지 송수신, 컨슈머 그룹 관리 등을 명령어로 수행할 수 있다.이 글에서는 실무에서 자주 사용하는 핵심 CLI 명령어를 정리해보겠다.

 

카프카 CLI 도구 위치

카프카를 설치하면 /bin 디렉토리에서 다양한 쉘 스크립트를 찾을수 있다.

# 카프카 설치 디렉토리로 이동
cd /path/to/kafka
# bin 디렉토리 확인
ls bin/

 

주요 CLI 도구

• kafka-topics.sh - 토픽관리
• kafka-console-producer.sh - 메세지 발행
• kafka-console-consumer.sh - 메세지 구독
• kafka-consumer-groups.sh - 컨슈머 그룹 관리
• kafka-configs.sh - 설정 관리
• kafka-log-dirs.sh - 로그 디렉토리 확인

Tip: PATH 환경변수에 카프카 bin 디렉토리를 추가하면 어디서든 명령어를 실행할 수 있다.

# .bashrc 또는 .zshrc에 추가
export PATH=$PATH:/path/to/kafka/bin

 

1. 토픽 관리(kafka-topics.sh)

토픽은 카프카에서 메세지를 저장하는 논리적인 단위다. RDBMS의 테이블과 유사한 개념이다.

토픽 목록 조회

kafka -topics.sh --bootstrap-server localhost:9092 --list

토픽 생성

# 기본 생성
kafka-topics.sh --bootstrap-server localhost:9092 \
	--create \
    --topic my-topic

# 파티션과 복제 계수 지정
kafka-topics.sh --bootstrap-server localhost:9092 \
	--create \
    --topic my-topic \
    --partitions 3 \
    --replication-factor 2

 

주요 옵션

• --partitions: 파티션 수(병렬 처리 성능에 영향)
• --replication-factor: 복제 계수 (데이터 내구성)
• --config: 토픽별 설정 (retention.ms 등)

토픽 상세 정보 조회

kafka-topics.sh --bootstrap-server localhost:9092 \
	--describe \
    --topic my-topic

출력 예시

Topic: my-topic	PartitionCount: 3	ReplicationFactor: 2
Topic: my-topic	Partition: 0	Leader: 1	Replicas: 1,2	Isr: 1,2
Topic: my-topic	Partition: 1	Leader: 2	Replicas: 2,3	Isr: 2,3
Topic: my-topic	Partition: 2	Leader: 3	Replicas: 3,1	Isr: 3,1

토픽 수정

# 파티션 수 증가(감소는 불가능)
kafka-topics.sh --bootstrap-server localhost:9092 \
	--alter \
    --topic my-topic \
    --partitions 5

토픽 삭제

kafka-topics.sh --bootstrap-server localhost:9092 \
	--delete \
    --topic my-topic

 

⚠️ 주의: 토픽 삭제는 복구가 불가능하므로 신중하게 사용해야 한다

 

2. 메세지 발행 (kafka-console-producer.sh)

콘솔 프로듀서를 사용하여 테스트 메세지를 빠르게 발행할 수 있다.

 

기본 메세지 발행

kafka-console-producer.sh --bootstrap-server localhost:9092 \
  --topic my-topic

실행 후 터미널에서 메시지를 입력하고 Enter를 누르면 메시지가 발행된다.

 

Key-Value 메세지 발행

kafka-console-producer.sh --bootstrap-server localhost:9092 \
	--topic my-topic \
    --property "parse.key=true" \
    --property "key.separator=:"

 

입력 예시

> user1:{"name":"John","age":30}
> user2:{"name":"Jane","age":25}

파일에서 메세지 발행

# messages.txt 파일 내용을 토픽으로 발행
cat messages.txt | kafka-console-producer.sh \
  --bootstrap-server localhost:9092 \
  --topic my-topic

 

3. 메세지 구독(kafka-console-consumer.sh)

콘솔 컨슈머를 사용하여 토픽의 메세지를 실시간으로 확인할 수 있다.

 

최신 메세지부터 구독

kafka-console-consumer.sh --bootstrap-server localhost:9092 \
	--topic my-topic

처음부터 모든 메세지 읽기

kafka-console-consumer.sh --bootstrap-server localhost:9092 \
	--topic my-topic
    --from-beginning

Key-Value 출력

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

컨슈머 그룹으로 구독

kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic my-topic \
  --group my-consumer-group \
  --from-beginning

파티션별 메세지 확인

# 특정 파티션만 구독
kafka-console-consumer.sh --bootstrap-server localhost:9092 \
	-- topic my-topic \
    -- partition 0 \
    -- from-beginning

 

Offset과 Timestamp 출력

kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic my-topic \
  --from-beginning \
  --property print.offset=true \
  --property print.timestamp=true

 

 

출력 예시

CreateTime:1704434400000	Offset:0	Hello Kafka!
CreateTime:1704434401000	Offset:1	This is a test message.

 

4. 컨슈머 그룹 관리 (kafka-consumer-groups.sh)

컨슈머 그룹의 상태와 Offset을 모니터링하고 관리할 수 있다.

컨슈머 그룹 목록 조회

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list

컨슈머 그룹 상세 정보

kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group my-consumer-group \
  --describe

 

출력 예시

GROUP           TOPIC      PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG  CONSUMER-ID     HOST        CLIENT-ID
my-consumer-group my-topic  0          100             105             5    consumer-1      /127.0.0.1  consumer-1
my-consumer-group my-topic  1          200             200             0    consumer-2      /127.0.0.1  consumer-2
my-consumer-group my-topic  2          150             160             10   consumer-1      /127.0.0.1  consumer-1

 

주요 필드 설명

• CURRENT-OFFSET: 현재 컨슈머가 읽은 위치
• LOG-END-OFFSET: 파티션의 마지막 메세지 위치
• LAG: 처리해야할 메세지 수

Offset 리셋

# 가장 처음으로 리셋
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group my-consumer-group \
  --topic my-topic \
  --reset-offsets --to-earliest \
  --execute

# 가장 최신으로 리셋
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group my-consumer-group \
  --topic my-topic \
  --reset-offsets --to-latest \
  --execute

# 특정 Offset으로 리셋
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group my-consumer-group \
  --topic my-topic:0 \
  --reset-offsets --to-offset 100 \
  --execute

 

⚠️ 주의: Offset 리셋은 컨슈머 그룹이 활성화되지 않은 상태에서만 가능하다

컨슈머 그룹 삭제

kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group my-consumer-group \
  --delete

 

5. 설정 관리 (kafka-configs.sh)

토픽, 브로커, 클라이언트의 동적 설정을 관리할 수 있다.

토픽 설정 조회

kafka-configs.sh --bootstrap-server localhost:9092 \
	--entity-type topics \
    --entity-name my-topic \
    --describe

 

토픽 설정 변경

# retention 시간 변경 (7일)
kafka-configs.sh --bootstrap-server localhost:9092 \
  --entity-type topics \
  --entity-name my-topic \
  --alter \
  --add-config retention.ms=604800000

# 최대 메시지 크기 변경
kafka-configs.sh --bootstrap-server localhost:9092 \
  --entity-type topics \
  --entity-name my-topic \
  --alter \
  --add-config max.message.bytes=2097152

설정 삭제

kafka-configs.sh --bootstrap-server localhost:9092 \
  --entity-type topics \
  --entity-name my-topic \
  --alter \
  --delete-config retention.ms

6. 로그 및 파티션 정보 (kafka-log-dirs.sh)

브로커의 로그 디렉토리와 파티션 크기 정보를 확인할 수 있다.

kafka-log-dirs.sh --bootstrap-server localhost:9092 \
  --describe \
  --broker-list 1,2,3

 

출력 예시

{
  "brokers": [
    {
      "broker": 1,
      "logDirs": [
        {
          "logDir": "/var/kafka-logs",
          "partitions": [
            {
              "partition": "my-topic-0",
              "size": 1048576,
              "offsetLag": 0
            }
          ]
        }
      ]
    }
  ]
}

 

7. 메세지 검색 및 분석

특정 시간대 메세지 조회

# 특정 timestamp부터 메시지 읽기 (Unix timestamp)
kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic my-topic \
  --property print.timestamp=true \
  --from-beginning \
  | awk '$1 > 1704434400000'

메세지 개수 세기

# 토픽의 전체 메시지 수 확인
kafka-run-class.sh kafka.tools.GetOffsetShell \
  --broker-list localhost:9092 \
  --topic my-topic \
  --time -1

메세지 샘플링

# 최근 10개 메시지만 조회
kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic my-topic \
  --max-messages 10

8. 실무 활용 팁

1. 브로커 연결 정보 간소화

매번 --bootstrap-server를 입력하는 게 번거롭다면 환경변수를 설정하세요

# .bashrc 또는 .zshrc에 추가
export KAFKA_BROKER="localhost:9092"

# 사용
kafka-topics.sh --bootstrap-server $KAFKA_BROKER --list

2. JSON 메세지 예쁘게 출력

kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic my-topic \
  --from-beginning \
  | jq '.'

3. 메세지 필터링

# "error"가 포함된 메시지만 출력
kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic my-topic \
  --from-beginning \
  | grep "error"

4. 성능 테스트

# Producer 성능 테스트
kafka-producer-perf-test.sh \
  --topic my-topic \
  --num-records 1000000 \
  --record-size 1024 \
  --throughput -1 \
  --producer-props bootstrap.servers=localhost:9092

# Consumer 성능 테스트
kafka-consumer-perf-test.sh \
  --topic my-topic \
  --messages 1000000 \
  --bootstrap-server localhost:9092

5. 로그 모니터링

# LAG이 큰 컨슈머 그룹 찾기
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --all-groups \
  --describe \
  | awk '$6 > 1000 {print $0}'

자주 사용하는 명령어 체크리스트

작업	명령어
토픽 목록	kafka-topics.sh --list
토픽 생성	kafka-topics.sh --create --topic [name]
토픽 상세정보	kafka-topics.sh --describe --topic [name]
메시지 발행	kafka-console-producer.sh --topic [name]
메시지 구독	kafka-console-consumer.sh --topic [name]
처음부터 읽기	--from-beginning 옵션 추가
컨슈머 그룹 목록	kafka-consumer-groups.sh --list
그룹 상세정보	kafka-consumer-groups.sh --describe --group [name]
Offset 리셋	kafka-consumer-groups.sh --reset-offsets

정리

카프카 CLI 명령어는 다음과 같은 상황에서 매우 유용하다

 

✅ 개발 및 테스트: 빠른 프로토타이핑과 디버깅
✅ 운영 모니터링: 실시간 메시지 흐름 확인
✅ 문제 해결: Lag 분석, Offset 리셋 등
✅ 자동화 스크립트: 배치 작업 및 CI/CD 파이프라인
✅ 학습: 카프카 동작 원리 이해

 

카프카 CLI 도구는 강력하면서도 직관적이다. 이 가이드를 참고하여 실무에서 카프카를 효과적으로 활용하길 빈다.