Kafka Broker, Cluster, Zookeeper

Kafka Broker

• Producer로부터 데이터를 전달받으면, 해당 토픽의 파티션에 데이터를 저장
• Consumer가 데이터를 요청하면, 해당 토픽의 파티션에 저장된 데이터를 전달
• 데이터는 파일 시스템에 저장
  • config/server.properties 의 log.dir 옵션에 정의한 디렉토리에 저장

ls /tmp/kafka-logs

__consumer_offsets-0   __consumer_offsets-2   __consumer_offsets-30  __consumer_offsets-41  __consumer_offsets-8       log-start-offset-checkpoint
__consumer_offsets-1   __consumer_offsets-20  __consumer_offsets-31  __consumer_offsets-42  __consumer_offsets-9       meta.properties
__consumer_offsets-10  __consumer_offsets-21  __consumer_offsets-32  __consumer_offsets-43  cleaner-offset-checkpoint  recovery-point-offset-checkpoint
__consumer_offsets-11  __consumer_offsets-22  __consumer_offsets-33  __consumer_offsets-44  hello.kafka-0              replication-offset-checkpoint
__consumer_offsets-12  __consumer_offsets-23  __consumer_offsets-34  __consumer_offsets-45  hello.kafka.2-0            test-0
__consumer_offsets-13  __consumer_offsets-24  __consumer_offsets-35  __consumer_offsets-46  hello.kafka.2-1            test3-0
__consumer_offsets-14  __consumer_offsets-25  __consumer_offsets-36  __consumer_offsets-47  hello.kafka.2-2            verify-test-0
__consumer_offsets-15  __consumer_offsets-26  __consumer_offsets-37  __consumer_offsets-48  hello.kafka.2-3            verify-test-2-0
__consumer_offsets-16  __consumer_offsets-27  __consumer_offsets-38  __consumer_offsets-49  hello.kafka.3-0            verify-test-3-0
__consumer_offsets-17  __consumer_offsets-28  __consumer_offsets-39  __consumer_offsets-5   hello.kafka.4-0
__consumer_offsets-18  __consumer_offsets-29  __consumer_offsets-4   __consumer_offsets-6   hello.kafka.4-1
__consumer_offsets-19  __consumer_offsets-3   __consumer_offsets-40  __consumer_offsets-7   hello.kafka.4-2

Kafka Broker가 파일 시스템에 저장한 데이터

• 파티션 단위로 데이터를 저장한다.
• log
  • 메시지와 메타데이터를 저장
• index
  • 메시지의 오프셋을 인덱싱한 정보를 담은 파일
• timeindex
  • 메시지에 포함된 timestamp값을 기준으로 인덱싱한 정보를 담은 파일

ls /tmp/kafka-logs/hello.kafka.2-0

-rw-r--r--. 1 ec2-user ec2-user      235 May  7 13:23 00000000000000000000.log
-rw-r--r--. 1 ec2-user ec2-user 10485760 May  7 13:01 00000000000000000000.index
-rw-r--r--. 1 ec2-user ec2-user 10485756 May  7 13:01 00000000000000000000.timeindex
-rw-r--r--. 1 ec2-user ec2-user        8 May  7 13:01 leader-epoch-checkpoint

Kafka Replication

• fault tolerant system (장애 허용 시스템)으로 동작하도록 하는 원동력
• Cluster로 묶인 Broker 중 일부에 장애가 발생하더라도 데이터를 유실하지 않고 안전하게 사용하기 위함

Controller

Cluster의 다수 Broker 중 한 대가 Controller 역할을 한다.

• 다른 Broker들의 상태를 체크
• 특정 Broker가 Cluster에서 빠지는 경우 , 해당 Broker에 존재하는 리더 파티션을 재분배
• 만약 Controller 역할을 하는 Broker에 장애가 생기면, 다른 Broker가 Controller 역할을 한다.

데이터 삭제

Kafka는 다른 메시징 플랫폼과 다르게 Consumer가 데이터를 가져가더라도 Topic의 데이터는 삭제되지 않는다.

• ❌ Consumer나 Producer가 데이터 삭제를 요청할 수 없다.
• 오직 Broker만이 데이터를 삭제할 수 있다.
• 데이터 삭제는 파일 단위(Log Segment)로 이루어짐
  • Log Segment는 데이터가 쌓이는 동안 파일 시스템으로 열려있다.
  • log.segment.bytes
    • 해당 설정값 용량에 도달했을 때, Log Segment 파일이 닫힌다.
  • log.segment.ms
  • 닫힌 Log Segment 파일은 log.retention.bytes 또는 log.retention.ms 설정값이 넘으면 삭제된다.
• 오래된 데이터를 압축하는 정책을 가져갈 수 있다. (Topic 압축 정책)

Offset Commit

• Consumer Group이 특정 Topic의 특정 Partition으로부터 어느 레코드까지 가져갔는지 확인할 수 있다.
• 커밋한 오프셋은 __consumer_offests 토픽에 저장한다.

Cordinator

Consumer Group의 상태를 체크하고 파티션을 Consumer와 매칭되도록 분배하는 역할을 한다.

• Consumer가 Consumer Group에서 빠지면 매칭되지 않은 파티션을 정상 동작하는 Consumer로 할당하여 끊임없이 데이터가 처리되도록 도와준다.
• 파티션을 Consumer로 재할당하는 과정을 Rebalance 라고 부른다.

Zookeeper

Kafka의 메타데이터를 관리한다.

bin/zookeper-shell.sh my-kafka:2181

ls /
[admin, brokers, cluster, config, consumers, controller, controller_epoch, isr_change_notification, latest_producer_id_block, log_dir_event_notification, zookeeper]

get /brokers/ids/0
{"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},"endpoints":["PLAINTEXT://ec2-3-36-77-126.ap-northeast-2.compute.amazonaws.com:9092"],"jmx_port":-1,"host":"ec2-3-36-77-126.ap-northeast-2.compute.amazonaws.com","timestamp":"1683464001941","port":9092,"version":4}

get /controller
{"version":1,"brokerid":0,"timestamp":"1683464002133"}

ls /brokers/topics
[__consumer_offsets, hello.kafka, hello.kafka.2, hello.kafka.3, hello.kafka.4, test, test3, verify-test, verify-test-2, verify-test-3]

ls /brokers
[ids, seqid, topics]

Zookeeper에 다수의 Kafka Cluster를 사용하는 방법

• Zookeeper는 서로 다른 znode에 Kafka Cluster들을 설정하면 된다.
• znode 란 Zookeper에서 사용하는 데이터 저장 단위이다.
  • 1개의 znode에는 n개의 하위 znode가 존재
  • 파일 시스템 혹은 트리 구조와 비슷
• server.properties 에 Zookepper 옵션 정의 예시
  • 파이프라인용 Kafka Cluster
    • zookeper.connect=localhost:2181/pipeline
  • 실시간 추천용 Kafka Cluster
    • zookeper.connect=localhost:2181/recommend

Topic

Topic은 Kafka에서 데이터를 구분하기 위해 사용하는 단위이다.

• Topic은 1개 이상의 Partition을 가진다.
• Partitiona에는 Producer가 보낸 데이터들을 저장
  • 이 데이터를 Record라고 부른다

Partition

Partition은 Kafka의 병렬처리의 핵심으로써 Consumer Group으로 묶인 Consumer들이 레코드를 병렬로 처리할 수 있도록 매칭된다.

Record

• 레코드는 다음과 같이 구성되어 있다.
  • 타임스탬프
  • 메시지 키
  • 메시지 값
  • 오프셋
  • 헤더
• Producer가 생성한 레코드가 Broker로 전송되면, 오프셋과 타임스탬프가 지정되어 저장된다.
• ❌ Broker에 한번 저장된 레코드는 수정할수 없다.
  • 로그 리텐션 기간 또는 용량에 따라서 삭제된다.
    • log.retention.ms
    • log.retention.bytes

타임스탬프

• Producer에서 해당 레코드가 생성된 시점(CreateTime)의 유닉스 타임이 설정된다.
• Producer가 레코드를 생성할 때 임의의 타임스탬프 값을 설정할 수 있다.

메시지 키

• 메시지 값을 순서대로 처리하거나 메시지 값의 종류를 나타내기 위해 사용
• 메시지 키의 해시값을 토대로 파티션을 지정하게 된다.
• 메시지 키가 null이면, Producer에 설정된 Partitioner에 따라서 파티션에 분배해서 저장한다.

메시지 값

• 실질적으로 처리할 데이터가 들어있따.
• 메시지 키와 메시지 값은 직렬화 되어 브로커로 전송
• 컨슈머가 이용할 때는 동일한 형태로 역직렬화를 수행해야한다.

헤더

• 레코드의 추가적인 정보를 담는 메타데이터 저장소 용도로 사용
• 키/값 형태로 데이터를 추가하여, 레코드의 속성 (Schema 버전 등)을 저장하여 Consumer에서 참조할 수 있다.