Kafka

Kafka 다운로드와 KRaft 모드 브로커 기동

먼저 최신 Kafka 릴리스를 다운로드해 압축을 푼다.

$ tar -xzf kafka_2.13-4.3.1.tgz
$ cd kafka_2.13-4.3.1

로컬 환경에는 Java 17 이상이 설치되어 있어야 한다. Kafka는 로컬 스크립트+다운로드한 파일로 실행하거나 Docker 이미지로 실행할 수 있다.

다운로드한 파일로 실행하는 경우, ZooKeeper 없이 KRaft 메타데이터 모드로 클러스터를 기동하려면 먼저 클러스터 UUID를 생성해야 한다.

$ KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"

다음으로 이 UUID를 이용해 로그 디렉토리를 포맷한다.

$ bin/kafka-storage.sh format --standalone -t $KAFKA_CLUSTER_ID -c config/server.properties

마지막으로 Kafka 서버(브로커)를 기동한다.

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

서버가 정상적으로 기동되면 기본적인 Kafka 환경이 준비되어 사용할 수 있는 상태가 된다.

Docker로 실행하기

로컬 스크립트 대신 Docker 이미지로도 Kafka를 띄울 수 있다. JVM 기반 공식 이미지를 쓰는 경우:

$ docker pull apache/kafka:4.3.1
$ docker run -p 9092:9092 apache/kafka:4.3.1

GraalVM 기반 네이티브 이미지를 쓰는 경우:

$ docker pull apache/kafka-native:4.3.1
$ docker run -p 9092:9092 apache/kafka-native:4.3.1

두 방식 모두 브로커의 기본 포트인 9092를 호스트에 매핑해 컨테이너 밖에서 접속할 수 있도록 한다.

토픽 생성과 상세 조회

Kafka는 이벤트(레코드/메시지)를 읽고 쓰고 저장하고 처리할 수 있게 해주는 분산 이벤트 스트리밍 플랫폼이다. 결제 트랜잭션, 모바일 기기의 위치 업데이트, 배송 주문, IoT/의료기기 센서 측정값 등이 이벤트의 예다. 이런 이벤트는 토픽에 조직되어 저장되므로, 첫 이벤트를 쓰기 전에 먼저 토픽을 만들어야 한다. 새 터미널 세션을 열고 다음을 실행한다.

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

Kafka의 모든 커맨드라인 도구는 인자 없이 실행하면 사용법을 보여준다. 예를 들어 describe 옵션으로 새로 만든 토픽의 파티션 수 같은 세부 정보를 확인할 수 있다.

$ bin/kafka-topics.sh --describe --topic quickstart-events --bootstrap-server localhost:9092
Topic: quickstart-events        TopicId: NPmZHyhbR9y00wMglMH2sg PartitionCount: 1
Topic: quickstart-events Partition: 0    Leader: 0   Replicas: 0 Isr: 0

출력에서 PartitionCount는 이 토픽이 파티션 1개로 생성되었음을(옵션 없이 create를 실행했을 때의 기본값) 보여주고, 두 번째 줄은 파티션 0의 리더 브로커가 ID 0번이며, 그 복제본(Replicas)과 ISR(In-Sync Replicas) 목록도 브로커 ID 0 하나뿐임을 보여준다 — 단일 브로커로 기동한 로컬 quickstart 환경이기 때문이다.

콘솔 프로듀서/컨슈머로 이벤트 쓰고 읽기

Kafka 클라이언트는 이벤트를 쓰거나 읽기 위해 네트워크를 통해 브로커와 통신한다. 브로커는 받은 이벤트를 필요한 만큼(영구히도 가능) durable하고 장애에 강하게 저장한다.

콘솔 프로듀서 클라이언트를 실행해 토픽에 이벤트 몇 개를 써 본다. 기본적으로 입력하는 한 줄 한 줄이 별도의 이벤트로 기록된다.

$ bin/kafka-console-producer.sh --topic quickstart-events --bootstrap-server localhost:9092
>This is my first event
>This is my second event

프로듀서 클라이언트는 언제든 Ctrl-C로 멈출 수 있다.

다른 터미널 세션을 열어 콘솔 컨슈머 클라이언트로 방금 만든 이벤트를 읽는다.

$ bin/kafka-console-consumer.sh --topic quickstart-events --from-beginning --bootstrap-server localhost:9092
This is my first event
This is my second event

컨슈머 클라이언트도 Ctrl-C로 멈출 수 있다. 프로듀서 터미널로 돌아가 이벤트를 더 써 보면 컨슈머 터미널에 즉시 나타나는 것을 확인할 수 있다. 이벤트는 durable하게 저장되므로 원하는 만큼, 원하는 수의 컨슈머로 반복해서 읽을 수 있다 — 다른 터미널을 열어 같은 컨슈머 명령을 다시 실행해도 처음부터 다시 읽힌다.

Kafka Connect로 외부 시스템과 데이터 주고받기

기존 시스템(관계형 데이터베이스, 전통 메시징 시스템 등)에 이미 대량의 데이터와 그 데이터를 쓰는 애플리케이션들이 있는 경우가 많다. Kafka Connect는 외부 시스템의 데이터를 Kafka로 지속적으로 수집하거나 반대로 내보낼 수 있게 해주는 도구다. 커넥터(connector)라는 확장 가능한 컴포넌트를 실행하는 방식으로 동작하며, 각 커넥터는 특정 외부 시스템과 상호작용하는 커스텀 로직을 구현한다. 그래서 기존 시스템을 Kafka와 통합하는 일이 매우 쉬워지며, 이미 커뮤니티가 제공하는 수백 개의 커넥터가 있어 대부분의 경우 직접 만들 필요가 없다.

이번 Quick Start에서는 파일에서 Kafka 토픽으로 데이터를 가져오는 간단한 커넥터와, 반대로 토픽에서 파일로 데이터를 내보내는 커넥터를 standalone 모드(단일 로컬 전용 프로세스)로 실행해본다.

먼저 Connect 워커 설정의 plugin.path 속성에 connect-file-4.3.1.jar를 추가해야 한다. 이 Quick Start에서는 설치 디렉토리에서 명령을 실행한다는 전제로 상대 경로를 쓰지만, 운영 환경에서는 절대 경로를 쓰는 것이 항상 바람직하다. config/connect-standalone.properties 파일을 열어 plugin.path 설정을 추가하거나 다음 명령으로 덧붙인다.

$ echo "plugin.path=libs/connect-file-4.3.1.jar" >> config/connect-standalone.properties

이어서 테스트용 시드 데이터를 만든다 (Linux/macOS).

$ echo -e "foo\nbar" > test.txt

Windows에서는 다음과 같이 두 줄로 작성한다.

$ echo foo > test.txt
$ echo bar >> test.txt

이제 standalone 모드로 커넥터 두 개를 실행한다. 설정 파일 세 개를 인자로 주는데, 첫 번째는 항상 Connect 프로세스 자체의 공통 설정(연결할 브로커, 데이터 직렬화 포맷 등)이고, 나머지 두 개는 각각 하나의 커넥터를 정의하는 설정(고유한 커넥터 이름, 인스턴스화할 커넥터 클래스, 그 밖의 필요한 설정)이다.

$ bin/connect-standalone.sh config/connect-standalone.properties config/connect-file-source.properties config/connect-file-sink.properties

Kafka에 함께 포함된 이 예제 설정 파일들은 앞서 기동해 둔 기본 로컬 클러스터 설정을 그대로 사용하며 커넥터 두 개를 만든다 — 하나는 입력 파일에서 줄 단위로 읽어 Kafka 토픽으로 게시하는 source 커넥터, 다른 하나는 Kafka 토픽에서 메시지를 읽어 출력 파일에 한 줄씩 쓰는 sink 커넥터다. 시작하는 동안 커넥터가 인스턴스화되었다는 로그 메시지들이 보인다. Connect 프로세스가 시작되면 source 커넥터는 test.txt에서 줄을 읽어 connect-test 토픽에 게시하기 시작하고, sink 커넥터는 connect-test 토픽에서 메시지를 읽어 test.sink.txt 파일에 쓰기 시작한다. 출력 파일 내용을 확인해 파이프라인 전체로 데이터가 잘 전달되었는지 검증할 수 있다.

$ more test.sink.txt
foo
bar

데이터는 connect-test라는 Kafka 토픽에도 저장되므로, 콘솔 컨슈머로 직접 확인하거나(또는 커스텀 컨슈머 코드로 처리) 볼 수 있다.

$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic connect-test --from-beginning
{"schema":{"type":"string","optional":false},"payload":"foo"}
{"schema":{"type":"string","optional":false},"payload":"bar"}

출력에서 보듯, 파일 커넥터가 만든 메시지에는 값(payload)뿐 아니라 스키마(schema) 정보가 함께 실려 있다. 커넥터는 데이터를 계속 처리하므로 파일에 줄을 더 추가하면 파이프라인을 타고 흘러가는 것을 볼 수 있다.

$ echo "Another line" >> test.txt

추가한 줄이 콘솔 컨슈머 출력과 sink 파일 양쪽에 나타나는 것을 확인할 수 있다.