Kakfa Topic, Producer, Consumer

Topic , Parition , parallel processing a kafka

• topic은 여러개의 parition로 이루어져 있고 producer가 보낸 message를 저장하게 된다.
• 즉 topic은 partition으로 구성된 일련의 로그 파일이다.
• Rdbms 의 partitioned table과 유사하다.
• topic은 key value구조이고 value는 어떤 타입의 메시지도 가능하다 (Json, String, Object, ProtoBuf등)

producer가 보낸 data는 topic에 offset이 순차적으로 증가하면서 저장이 된다. 계속 append 만되는것이다.
중간에 껴서 update하고 그런것이 없다.

개별 파티션은 정렬되고, immutable 일련의 레코드로 구성된 로그 메시지이다.
개별 파티션은 다른 파티션과 완전히 독립적이다.

replication factor 2 인 경우에 각각 broker에서 같은 topic A에 대한 partition을 서로 복사를 해놓은다.

예를들어 , 2번 broker server down이 일어나면 1번 broker의 복제영역에 복제되었던 partition#1 이 leader가 되면서 진행된다.

---

kafka-topics command

test_topic_02를 partition 3 으로 생성했을 때 --decsribe로 봤을때 결과이다.

현재는 vm에 broker가 1개 띄워져있기에 replication 은 불가능하다.

kafka topic을 delete를 시전하면 바로 지워지지 않고 marking이 된후에 시간이 지나면 지워지게 된다.

---

Producer , Consumer

producer 가 보내는 message는 어떤 paritiion으로 보낼지 전략적으로 결정이된다. 
record는 위와같이 여러개로 구성이 된다.

topic == 주소
value == 실제 message의 body를 의미하는것 같다. 
이 2가지는 must 로 존재해야한다.

---

kafka console producer, kafka console consumer

 

auto.offset.reset 의 값은 consumer가 topic에 처음 접속하여 message를 가져올 때 가장 오래된 처음 offset 부터 (earliset) 가져올 것인지, 가장 최근인 마지막 offset 부터 가져올 것인지를 설정하는 파라미터

earliest ,latest 가 존재한다. 기본이 latest이다.

kafka console consumer 명령어를 사용할 때 --from-beginning을 사용해야하만 auto.offset.reset이 earlist로 지정됨

---

Producer Serializer의 이해

 

producer record는 항상 serialize를 통해서 byte[] 배열로 되서 broker의 topic의 partition으로 전송 , 실제 저장되는것도 byte[]

consumer는 deserialize를 수행해서 record를 복구함

String, Short, Integer, Long, Double, Bytes [ Serializer ] 들을 kafka에서 client app에서 쓸 수 있도록 제공한다.

---

key 값을 가지는 producer record

key값을 가지지 않는 경우에는 partitioner가 Round robin, stick parition 등의 전략으로 알아서 어디로 전송할지 정해주게 된다.

RR같은 경우 topic이 여러개의 파티션을 가질때 consumer에서는 전송 순서가 보장되지 않게 될 수 있다.

topic의 partition을 1개만 가져가면 전송순서가 보장이된다. 이게 분산시스템이니 partition이 여러개 일때도 전송순서를 보장하기 위해 consumer가 소비하려 할 때마다 모든 partition을 뒤져서 순서를 맞추는것은 성능을 저하 시키게 된다.

 

key값을 가지는 경우에는 특정 key값을 가지는 메시지는 특정 파티션으로 고정되어 전송이 된다. 즉 메세지의 순서를 보장하게 된다.

message key값은 message의 식별자는 아니다. 예를들면 user id가 될 수 도 있다.

순서가 보장됨으로 consume group에서 각 consumer가 각 parition을 읽게 할 수 있다.

key값을 준 상태로 producer가 record를 보내고 consumer가 소비한 예시이다.

key값을 주지 않으면 batch.size 만큼 채워지기 전까지는 sticky partitioner처럼 동작을 하게된다.

RR로 동작하게 만들려면

위와 같이 옵션을 줘야 key값 없이도 RR로 동작하게 된다.

• If not set, the default partitioning logic is used. This strategy send records to a partition until at least batch.size bytes is produced to the partition. It works with the strategy:If no partition is specified but a key is present, choose a partition based on a hash of the key.If no partition or key is present, choose the sticky partition that changes when at least batch.size bytes are produced to the partition.

 

---

Consumer Group , Consumer

모든 Consumer들은 단 하나의 Consumer Group 에 소속되어야 하며, Consumer Group은 1개 이상의 Consumer를 가질 수 있음.
파티션의 레코드들은 단 하나의 Consumer에만 할당된다.
(하나의 partition 이 동시에 consumer 4,,5 막 이렇게 가지 않는다.)

consume할 때 partition 들간의 순서가 보장이 안되어서 읽을 수 있다. producer 의 paritioner 처럼 consumer의 assigner가 존재한다.

consumer group에 consumer가 추가 될때마다 rebalancing이 일어나서 담당하는 partition이 달라지게 된다.

consumer group끼리는 독립적이다.

kafka consumer group --describe 명령어로 어떤 partition이 어떤 consumer랑 mapping이 되어있는지 확인 할 수 있다.

consumer가 토픽을 읽는 정보는 consumer group 단위로 관리되고 __consumer_offsets 에 해당 내용을 저장한다.
동일한 consumer group 내에서는 서로 다른 consumer 들의 중복 읽기를 허락하지 않는다

 

---

Kafka Config 구분 및 이해

Broker와 Topic 레벨 Config

• Kafka 서버에서 설정되는 Config
• Topic의 Config 값은 Broker 레벨에서 지정한 Config를 기본으로 설정하며 별도의 Topic 레벨 Config를 설정할 경우 이를 따름
• 보통 server.properties 에 있는 Config 는 변경시 Broker 재기동이 필요한 Static Config이며, Dynamic Config는 kafka-configs 를 이용하여 동적으로 config 변경 가능

Producer 와 Consumer 레벨 Config

• Kafka client에서 설정되는 Config
• Client 레벨에서 설정되므로 server.properties에 존재하지 않고, kafka-config로 수정할 수 없으며 client 수행할때마다 설정할 수 있음

kafka-config 명령어 는 entity type (broker, topic) , entity name을 인자로 받은 다음에 이 객체에 적용된 설정들을 list 하거나 control 할 때 필요한 명령어이다.

kafka-dump-log 명령어로 data/kafka-logs 안에있는 특정 로그를 읽어올 수 있다.